scala - Flink Scala NotInferedR in scala Type mismatch MapFunction[Tuple2[Boolean,Row],InferedR]

Question

我正在尝试在 MapFunction 的 flink 中将 Tuple2[Boolean,Row] 转换为 Row ，但失败并出现错误。

当我尝试运行时，我得到另一个错误。

代码我想要做什么

    val data = kinesis.map(mapFunction)
    tEnv.registerDataStream("transactions", data, "cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long")
    val query = "SELECT distinct cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long FROM transactions where cc_num not in ('cc_num') "
    val table = tEnv.sqlQuery(query)

    val mymapFunction: MapFunction[tuple.Tuple2[Boolean, Row],AnyVal] =
      new MapFunction[tuple.Tuple2[Boolean, Row],AnyVal]() {
        override def map(t: tuple.Tuple2[Boolean, Row]): Row = {
          t.f1
        }
        /*override def map(t: tuple.Tuple2[Boolean, Row], collector: Collector[Object]): Unit = {
          collector.collect(t.f1)
        }
      */
  }

    tEnv.toRetractStream(table, classOf[org.apache.flink.types.Row]).map(mymapFunction)
      .writeAsCsv("/Users/kalit_000/Downloads/FlinkStreamAndSql/src/main/resources/csvOut8",
      FileSystem.WriteMode.OVERWRITE,"\n","|")

谢谢斯里

Answer 1

在编写 Flink scala 应用程序时进行了以下更改，它具有不同的导入集。

从：-

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.api.java.typeutils.TupleTypeInfo
import org.apache.flink.table.sinks.RetractStreamTableSink

tEnv.toRetractStream(table, classOf[org.apache.flink.types.Row])
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv: org.apache.flink.table.api.java.StreamTableEnvironment = TableEnvironment.getTableEnvironment(env)

至：-

 import org.apache.flink.api.scala._
 import org.apache.flink.table.api.scala._
 import org.apache.flink.streaming.api.scala.{DataStream, 
 StreamExecutionEnvironment}
 import org.apache.flink.table.api.scala.StreamTableEnvironment
 import org.apache.flink.api.common.typeinfo.TypeInformation

 table.toRetractStream(TypeInformation.of(classOf[Row]))
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

如果有人想要，这里有完整的代码：-

package com.aws.examples.kinesis.consumer.transactionExampleScala

import java.util.Properties
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
import com.aws.examples.kinesis.consumer.TransactionExample.TransactionJsonClass
import com.google.gson.Gson
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.types.Row
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer
import org.apache.flink.streaming.connectors.kinesis.config.{AWSConfigConstants, ConsumerConfigConstants}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
import java.sql.{DriverManager, Time}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.core.fs.{FileSystem, Path}

object TransactionScalaTest {

  def main(args: Array[String]): Unit = {

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //env.enableCheckpointing(10000)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // Get AWS credentials
    val credentialsProvider = new DefaultAWSCredentialsProviderChain
    val credentials = credentialsProvider.getCredentials

    // Configure Flink Kinesis consumer
    val consumerConfig = new Properties
    consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1")
    consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, credentials.getAWSAccessKeyId)
    consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, credentials.getAWSSecretKey)
    consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON")

    // Create Kinesis stream
    val kinesis = env.addSource(new FlinkKinesisConsumer("credittransactions3", new SimpleStringSchema(), consumerConfig))

    val mapFunction: MapFunction[String, (String, String, String, String, String, String, String, String, String, String)] =
      new MapFunction[String, (String, String, String, String, String, String, String, String, String, String)]() {

        override def map(s: String): (String, String, String, String, String, String, String, String, String, String) = {

          val data = new Gson().fromJson(s, classOf[TransactionJsonClass])

          val csvData = data.getCc_num + "," +
            data.getFirst + "," +
            data.getLast + "," +
            data.getTrans_num + "," +
            data.getTrans_time + "," +
            data.getCategory + "," +
            data.getMerchant + "," +
            data.getAmt + "," +
            data.getMerch_lat + "," +
            data.getMerch_long

          //println(csvData)

          val p: Array[String] = csvData.split(",")
          var cc_num: String = p(0)
          var first: String = p(1)
          var last: String = p(2)
          var trans_num: String = p(3)
          var trans_time: String = p(4)
          var category: String = p(5)
          var merchant: String = p(6)
          var amt: String = p(7)
          var merch_lat: String = p(8)
          var merch_long: String = p(9)

          val creationDate: Time = new Time(System.currentTimeMillis())
          return (cc_num, first, last, trans_num, trans_time, category, merchant, amt, merch_lat, merch_long)
        }
      }


    val data = kinesis.map(mapFunction)

    tEnv.registerDataStream("transactions", data, 'cc_num, 'first_column, 'last_column, 'trans_num,
      'trans_time, 'category_column, 'merchant_column, 'amt_column, 'merch_lat, 'merch_long)
    //tEnv.registerDataStream("transactions", data, "cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long")
    val query = "SELECT distinct cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long FROM transactions where cc_num not in ('cc_num') "
    val table = tEnv.sqlQuery(query)

    table.toRetractStream(TypeInformation.of(classOf[Row]))
      .map(_._2)
      .writeAsText("/Users/kalit_000/Downloads/FlinkStreamAndSql/src/main/resources/csvOut125", FileSystem.WriteMode.OVERWRITE)

    table.printSchema()

    table.toRetractStream(TypeInformation.of(classOf[Row])).print()

    env.execute()
  }
}

聚甲醛：-

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>FlinkStreamAndSql</groupId>
    <artifactId>FlinkStreamAndSql</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <!-- see http://davidb.github.com/scala-maven-plugin -->
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.1.3</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.13</version>
                <configuration>
                    <useFile>false</useFile>
                    <disableXmlReport>true</disableXmlReport>
                    <!-- If you have classpath issue like NoDefClassError,... -->
                    <!-- useManifestOnlyJar>false</useManifestOnlyJar -->
                    <includes>
                        <include>**/*Test.*</include>
                        <include>**/*Suite.*</include>
                    </includes>
                </configuration>
            </plugin>

            <!-- "package" command plugin -->
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4.1</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
    <dependencies>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.derby</groupId>
            <artifactId>derby</artifactId>
            <version>10.13.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_2.11</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala_2.11</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java</artifactId>
            <version>1.8.1</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.11</artifactId>
            <version>1.8.1</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>1.8.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.11</artifactId>
            <version>1.8.1</version>
        </dependency>

       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-scala_2.11</artifactId>
           <version>1.8.1</version>
       </dependency>

       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-streaming-scala_2.11</artifactId>
           <version>1.8.1</version>
       </dependency>

               <dependency>
                   <groupId>org.apache.flink</groupId>
                   <artifactId>flink-connector-kinesis_2.11</artifactId>
                   <version>1.8.0</version>
               </dependency>

               <dependency>
                   <groupId>org.apache.flink</groupId>
                   <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
                   <version>1.8.1</version>
               </dependency>

               <dependency>
                   <groupId>com.amazonaws</groupId>
                   <artifactId>amazon-kinesis-client</artifactId>
                   <version>1.8.8</version>
               </dependency>

               <dependency>
                   <groupId>com.amazonaws</groupId>
                   <artifactId>aws-java-sdk-kinesis</artifactId>
                   <version>1.11.579</version>
               </dependency>

               <dependency>
                   <groupId>commons-dbcp</groupId>
                   <artifactId>commons-dbcp</artifactId>
                   <version>1.2.2</version>
               </dependency>
               <dependency>
                   <groupId>com.google.code.gson</groupId>
                   <artifactId>gson</artifactId>
                   <version>2.1</version>
               </dependency>

               <dependency>
                   <groupId>commons-cli</groupId>
                   <artifactId>commons-cli</artifactId>
                   <version>1.4</version>
               </dependency>

               <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-csv -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-csv</artifactId>
            <version>1.7</version>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-compress</artifactId>
            <version>1.4.1</version>
        </dependency>

        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>dynamodb-streams-kinesis-adapter</artifactId>
            <version>1.4.0</version>
        </dependency>

        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>dynamodb-streams-kinesis-adapter</artifactId>
            <version>1.4.0</version>
        </dependency>

        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk</artifactId>
            <version>1.11.579</version>
        </dependency>

    </dependencies>

</project>