从 kafka 主题激发流数据并写入外部路径中的文本文件答案

【问题标题】：Spark streaming data from kafka topic and write into the text files in external path从 kafka 主题激发流数据并写入外部路径中的文本文件
【发布时间】：2019-04-25 19:33:31
【问题描述】：

我想从 kafka 主题中读取数据并按键值分组，然后写入文本文件..

public static void main(String[] args) throws Exception {
        SparkSession spark=SparkSession
                .builder()
                .appName("Sparkconsumer")
                .master("local[*]")
                .getOrCreate();
        SQLContext sqlContext = spark.sqlContext();
        SparkContext context = spark.sparkContext();
        Dataset<Row>lines=spark
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9092")
                .option("subscribe","test-topic")
                .load();
    Dataset<Row> r=  lines.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
        r.printSchema();
        r.createOrReplaceTempView("basicView");
        sqlContext.sql("select * from basicView")
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .writeStream()
        .outputMode("append")
        .format("console")
        .option("path","usr//path")
        .start()
        .awaitTermination();

【问题讨论】：

format(”console“) 写入控制台/标准输出而不是文件。改用 json 或 csv 作为参数
您遇到的错误是什么？
不知道为什么你会使用 Spark 来做这个
@AydinK。我知道那个，但是当我运行这个程序时，它会创建文件夹，并且该文件夹中没有任何内容.. 那是错误
@cricket_007 我想从 Kafka 主题流式传输数据..

标签： java apache-spark apache-kafka spark-structured-streaming

【解决方案1】：

以下几点在您的代码中具有误导性：

要从 Kafka 读取并写入文件，您不需要 SparkContext 或 SQLContext，
您将 key 和 value 两次转换为字符串，
如果要将数据存储到文件中，则输出查询的 format 不应为 console。

可以在Spark Structured Streaming + Kafka Integration Guide和Spark Structured Streaming Programming Guide中查找示例

public static void main(String[] args) throws Exception {

  SparkSession spark = SparkSession
    .builder()
    .appName("Sparkconsumer")
    .master("local[*]")
    .getOrCreate();

  Dataset<Row> lines = spark
    .readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe","test-topic")
    .load();

  Dataset<Row> r = lines
    .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    // do some more processing such as 'groupBy'
    ;

  r.writeStream
    .format("parquet")        // can be "orc", "json", "csv", etc.
    .outputMode("append")
    .option("path", "path/to/destination/dir")
    .option("checkpointLocation", "/path/to/checkpoint/dir")
    .start()
    .awaitTermination();

【讨论】：