Spark结构化流到kudu上下文

Question

我想阅读kafka主题，然后通过火花流将其写入kudu表。

我的第一个方法

// sessions and contexts
val conf = new SparkConf().setMaster("local[2]").setAppName("TestMain")
val sparkSession = SparkSession.builder().config(conf).getOrCreate()
val sparkContext = sparkSession.sparkContext
val kuduContext = new KuduContext("...", sparkContext);
// structure
val schema: StructType = StructType(
StructField("userNo", IntegerType, true) ::
StructField("bandNo", IntegerType, false) ::
StructField("ipv4", StringType, false) :: Nil);
// kudu - prepare table
kuduContext.deleteTable("test_table");
kuduContext.createTable("test_table", schema, Seq("userNo"), new CreateTableOptions()
.setNumReplicas(1)
.addHashPartitions(List("userNo").asJava, 3))
// get stream from kafka
val parsed = sparkSession
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "...")
.option("startingOffsets", "latest")
.option("subscribe", "feed_api_band_get_popular_post_list")
.load()
.select(from_json(col("value").cast("string"), schema).alias("parsed_value"))
// write it to kudu
kuduContext.insertRows(parsed.toDF(), "test_table");

现在它抱怨

Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
kafka
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$analysis$UnsupportedOperationChecker$$throwError(UnsupportedOperationChecker.scala:297)
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$$anonfun$checkForBatch$1.apply(UnsupportedOperationChecker.scala:36)

我的第二种方法

我似乎更改了代码以使用传统的KafkaUtils.createDirectStream

KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
).foreachRDD(rdd => {
rdd.foreach(record => {
// write to kudu.............
println(record.value());
})
});
ssc.start();
ssc.awaitTermination();

那么，哪一种方法是正确的呢？或者有没有办法让它从第一个方法开始运行？

Spark版本是2.2.0。

Answer 1

这两种方法似乎都是正确的。第一种是使用Spark结构化的流式处理方式，其中数据是在表格的基础上附加的。第二种方法是通过传统的DStream做事方式

Answer 2

我相信目前没有Kudu支持在Spark结构化流中使用KuduContext。我也遇到了类似的问题，不得不重新使用传统的Kudu客户端并实现ForeachWriter[Row]类。我使用了这里的例子，并能够实现一个解决方案。

Answer 3

正如您已经从错误中看到的那样，第一种方法是不正确的，它非常清楚：Queries with streaming sources must be executed with writeStream.start()。这只适用于批量。

第二个使用DStream，因此不是结构化流。

有第三种和第四种方法。

从Kudu 1.9.0开始，结构化流媒体得到支持，该问题得到修复，并按预期使用：

parsed
.writeStream
.format("kudu")
.option("kudu.master", kuduMaster)
.option("kudu.table", tableName)
.option("kudu.operation", operation)
.start()

请注意，如果您使用Cloudera，此方法将仅适用于cdh6.2.0及以上版本：

<!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2 -->
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-spark2_2.11</artifactId>
<version>1.9.0-cdh6.2.0</version>
<scope>test</scope>
</dependency>

我的解决方案是查看SparkContext中的代码，看看kuduContext.insertRows(df, table)和其他方法的作用，并创建一个ForeachWriter[Row]:

val kuduContext = new KuduContext(master, sparkContext)
parsed
.toDF()
.writeStream
.foreach(new ForeachWriter[Row] {
override def open(partitionId: Long, version: Long): Boolean =
kuduContext.tableExists(table)
override def process(value: Row): Unit = {
val kuduClient = kuduContext.syncClient
val kuduSession = kuduClient.newSession()
kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND)
kuduSession.setIgnoreAllDuplicateRows(ignoreDuplicates)
val kuduTable = kuduClient.openTable(kuduSinkConfiguration.table)
val operation = getOperationFunction(kuduTable) //get the kuduTable.newInsert(), newUpsert(), etc.
kuduSession.setIgnoreAllDuplicateRows(ignoreDuplicates)
val row = operation.getRow
row.add("userNo", value.getAs[Int]("userNo"))
row.add("bandNo", value.getAs[Int]("bandNo"))
row.add("ipv4", value.getAs[String]("ipv4"))
kuduSession.apply(operation)
kuduSession.flush()
kuduSession.close()
}
override def close(errorOrNull: Throwable): Unit = Unit
})
.start()

Answer 4

我们还可以使用Spark版本2.2.0和cloudera版本CDH 5.14将结构化流数据加载到Kudu表中。您只需要下载与CDH6.2对应的spark-kudu-2.2.11 jar，并在您的spark-submit命令中将其作为jar传递。这将识别下面语句中的kudu格式，并轻松加载数据帧。

解析的.writeStream.格式("kudu").option("kudu.master"，kuduMaster).option("kudu.table"，tableName).option("kudu.operation"，操作).start()

JAR可以从以下位置下载：https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2_2.11/1.10.0-cdh6.3.2

Spark提交声明：

spark2-submit--master local[*]--deploy mode client--jars spark-sql-kafka-0-10_2.12.2.0.jar，kafka-clients-0.10.0.0.jar，spark-streaming-kafka0-10_2.12-2.0.jar，kudu-spark2.2.11.10.0-cdh6.3.2.jar，kudu-client-1.0.0-cdh6.3.2.jar/path_of_python_code/rdd-stream-read.py

注意-Kudu客户端是可选的。可能必须与集群部署模式一起使用。

使用的writestream语句：

query=dfCols.writeStream.format("kudu").option("kudu.master"，"host:7051，host:7501").option("kudu.table"，"impala:：db.kudu_table_name").ortion("kudu.operation"，"upstart").options("checkpointLocation"，"file:///path_of_dir/checkpoint/").start()