试图理解结构化流

Question

我是新来的Apache火花，并试图理解结构化流与Apache卡夫卡在斯卡拉，但没有工作在我的青睐，直到现在基本上我想发送JSON从卡夫卡过程它使用火花结构化流并发回到卡夫卡。我试着在网站上给出的例子，但它不工作。

这里是我的代码：

import org.apache.spark.sql._ 
import org.apache.spark.sql.functions._ 
import org.apache.spark.sql.types.StructType 
import org.apache.spark.sql.types._ 
import org.apache.spark.sql.streaming.{OutputMode, Trigger} 
object dataset_kafka { 
    def main(args: Array[String]): Unit = { 

    val spark = SparkSession 
     .builder() 
     .appName("kafka-consumer") 
     .master("local[*]") 
     .getOrCreate() 

    import spark.implicits._ 

    spark.sparkContext.setLogLevel("WARN") 

    val df = spark 
     .readStream 
     .format("kafka") 
     .option("kafka.bootstrap.servers", "172.21.0.187:9093") 
     .option("subscribe", "test") 
     .load() 

     df 
     .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") 
     .writeStream 
     .format("kafka") 
     .trigger(Trigger.ProcessingTime("5 seconds")) 
     .option("kafka.bootstrap.servers", "172.21.0.187:9093") 
     .option("topic", "test1") 
     .option("checkpointLocation", "/home/hduser/Desktop/tempo") 
     .start() 
     .awaitTermination() 
    } 
} 

与我要去哪里不对任何帮助？

我以这种格式发送卡夫卡JSON：

{"schema":"Hiren","payload":"123"} 

Answer 1

我最近出版的专注于Apache的斯巴克的结构化数据流的技术资产的选集。其中一个博客侧重于结构化流式传输：即，使用结构化流式API将数据从Kafka流式传输到Spark中。

我会敦促你偷看它。请注意，只有在Spark 2.2中支持回写Kafka。

https://databricks.com/blog/2017/08/24/anthology-of-technical-assets-on-apache-sparks-structured-streaming.html