无法通过在使用 scala 的 spark Streaming 中从 kafka 主题中获取数据来编写 csv 文件

Question

我正在尝试使用以下代码从 Kafka 主题中读取数据：

object Main {
  def main(args: Array[String]) {
    val sparkSession = createSparkSession()
    val df = sparkSession.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test").option("startingOffsets", "earliest").load()
    val df1 = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

    df1.writeStream.format("parquet").option("format","append").option("checkpointLocation", "/home/krishna/Downloads/kafka_2.12-2.0.0/delete").option("path", "/home/krishna/Downloads/kafka_2.12-2.0.0/abc").option("truncate", "false").outputMode("append").start()
  }
}

当我使用以下行时：

df1.writeStream
  .format("console")
  .option("truncate","false")
  .start()
  .awaitTermination()

然后输出将显示在控制台上。

但问题是当我在代码行下面替换上面的行时：

df1.writeStream
  .format("csv")
  .option("format","append")
  .option("checkpointLocation", "/home/krishna/Downloads/kafka_2.12-2.0.0/delete")
  .option("path", "/home/krishna/Downloads/kafka_2.12-2.0.0/abc")
  .option("truncate", "false")
  .outputMode("append")
  .start()

那么输出不会以 CSV 格式保存。仅创建 abc 文件夹，并在其中创建元数据文件夹，但其中没有 CSV 文件。

我无法理解如果 o/p 成功显示在控制台上，那么为什么它没有以 csv、parquet 或文本的形式保存在文件中。
示例输出：

------------------
| key  | value   |
------------------
| null | abc     |
| null | 123     |
|-----------------

依赖：

<dependencies>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>2.4.5</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>2.4.5</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.12</artifactId>
    <version>2.4.5</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    <version>2.4.5</version>
    <scope>provided</scope>
  </dependency>
</dependencies>

Answer 1

在控制台中，您使用的是df，而对于csv，您使用的是df1。

大部分代码对我来说都很好。

试试这个。

df.writeStream 
    .format("csv")
    .option("format", "append")
    .trigger(processingTime = "5 seconds")
    .option("checkpointLocation", "/home/krishna/Downloads/kafka_2.12-2.0.0/delete")
.option("path", "/home/krishna/Downloads/kafka_2.12-2.0.0/abc")
    .outputMode("append")
    .start()

Answer 2

试试这个：

df.writeStream
.outputMode(OutputMode.Append())
.format("csv")
.option("checkpointLocation", "/home/krishna/Downloads/kafka_2.12-2.0.0/delete")
.option("path", "/home/krishna/Downloads/kafka_2.12-2.0.0/abc/")
.start()

您可以使用格式类型：com.databricks.spark.csv

Answer 3

我已经使用 Spark 2.4.5 测试了以下代码，它会根据需要生成 csv 文件：

 val sparkSession = SparkSession.builder()
    .appName("myAppName")
    .master("local[*]")
    .getOrCreate()

  val df = sparkSession.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "test")
    .option("startingOffsets", "earliest")
    .load()

  val df1 = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

  df1.writeStream
    .format("csv")
    .outputMode("append")
    .option("set", ",")
    .option("checkpointLocation", "/home/krishna/Downloads/kafka_2.12-2.0.0/delete")
    .option("path", "/home/krishna/Downloads/kafka_2.12-2.0.0/abc.csv")
    .start()
    .awaitTermination()

此代码将创建一个名为 abc.csv 的文件夹。根据 Sparksession 的并行性（由 spark.default.parallelism 配置），您会发现与分区一样多的 csv 文件。文件数反映了写出时 DataFrame 中的分区数。如果您在此之前对其进行重新分区，您最终会得到不同数量的文件。

在我的情况下，分区是2，所以我在相应的文件夹中得到了这个输出：

> ~/abc.csv$ ll
total 28
drwxrwxr-x  3 x x 4096 Apr 18 17:01 ./
drwxr-xr-x 50 x x 4096 Apr 18 17:01 ../
-rw-r--r--  1 x x    8 Apr 18 17:00 part-00000-77250d4a-e3af-46ef-b572-5476a3d075dd-c000.csv
-rw-r--r--  1 x x    4 Apr 18 17:00 part-00000-82a76a8c-5977-4891-be36-1c2dc6837fb1-c000.csv