如何将KAFKA的单个主题流式传输，通过键过滤到HDF的多个位置？

Question

我不是要在多个HDFS位置上流式传输数据，该位置通过键过滤。因此，下面的代码不起作用。请帮助我找到编写此代码的正确方法

    val ER_stream_V1 = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", configManager.getString("Kafka.Server"))
        .option("subscribe", "Topic1")
        .option("startingOffsets", "latest")
        .option("failOnDataLoss", "false")
        .load()
val ER_stream_V2 = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", configManager.getString("Kafka.Server"))
.option("subscribe", "Topic1")
        .option("startingOffsets", "latest")
        .option("failOnDataLoss", "false")
        .load()

        ER_stream_V1.toDF()
        .select(col("key"), col("value").cast("string"))
        .filter(col("key")==="Value1")
        .select(functions.from_json(col("value").cast("string"), Value1Schema.schemaExecution).as("value")).select("value.*")
        .writeStream
        .format("orc")
        .option("metastoreUri", configManager.getString("spark.datasource.hive.warehouse.metastoreUri"))
        .option("checkpointLocation", "/tmp/teststreaming/execution/checkpoint2005")
        .option("path", "/tmp/test/value1")
        .trigger(Trigger.ProcessingTime("5 Seconds"))
        .partitionBy("jobid")
        .start()

        ER_stream_V2.toDF()
        .select(col("key"), col("value").cast("string"))
        .filter(col("key")==="Value2")
        .select(functions.from_json(col("value").cast("string"), Value2Schema.schemaJobParameters).as("value"))
        .select("value.*")
        .writeStream
        .format("orc")
        .option("metastoreUri", configManager.getString("spark.datasource.hive.warehouse.metastoreUri"))
        .option("checkpointLocation", "/tmp/teststreaming/jobparameters/checkpoint2006")
        .option("path", "/tmp/test/value2")
        .trigger(Trigger.ProcessingTime("5 Seconds"))
        .partitionBy("jobid")
        .start()

Answer 1

您不需要两个读者。创建一个并过滤两次。您可能还需要考虑startoffsets作为最早的以读取现有主题数据

。

val ER_stream = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", configManager.getString("Kafka.Server"))
    .option("subscribe", "Topic1")
    .option("startingOffsets", "latest")  // maybe change?
    .option("failOnDataLoss", "false")
    .load()
    .toDF()
    .select(col("key").cast("string").as("key"), col("value").cast("string"))

val value1Stream = ER_stream
    .filter(col("key") === "Value1")
    .select(functions.from_json(col("value"), Value1Schema.schemaExecution).as("value"))
    .select("value.*")

val value2Stream = ER_stream
    .filter(col("key") === "Value2")
    .select(functions.from_json(col("value"), Value2Schema.schemaJobParameters).as("value"))
    .select("value.*")

value1Stream.writeStream.format("orc")
    ...
    .start()

value2Stream.writeStream.format("orc")
    ...
    .start()