使用 Spark Structured Streaming 从多个 Kafka 主题读取并写入不同接收器的最佳方式是什么？

Question

我正在尝试编写一个 Spark Structured Streaming 作业，该作业从多个 Kafka 主题（可能是 100 个）中读取，并根据主题名称将结果写入 S3 上的不同位置。我已经开发了这个 sn-p 的代码，它当前从多个主题中读取并将结果输出到控制台（基于循环）并且它按预期工作。但是，我想了解性能影响是什么。这是推荐的方法吗？不建议有多个 readStream 和 writeStream 操作吗？如果是，推荐的方法是什么？

my_topics = ["topic_1", "topic_2"]

for i in my_topics:
    df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", bootstrap_servers) \
        .option("subscribePattern", i) \
        .load() \
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

    output_df = df \
        .writeStream \
        .format("console") \
        .option("truncate", False) \
        .outputMode("update") \
        .option("checkpointLocation", "s3://<MY_BUCKET>/{}".format(i)) \
        .start()

Answer 1

以下方法的优点。

1. 通用
2. 多线程，所有线程将单独工作。
3. 易于维护代码并为任何问题提供支持。
4. 如果一个主题失败，对生产中的其他主题没有影响。你只需要专注于失败的那个。
5. 如果您想提取特定主题的所有数据，您只需停止该主题的作业，更新或更改配置并重新启动相同的作业。

注意 - 下面的代码不是完整的通用代码，您可能需要更改或调整下面的代码。

topic="" // Get value from input arguments
sink="" // Get value from input arguments

df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", bootstrap_servers) \
        .option("subscribePattern", topic) \
        .load() \
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

    output_df = df \
        .writeStream \
        .format("console") \
        .option("truncate", False) \
        .outputMode("update") \
        .option("checkpointLocation", sink) \
        .start()        

以下方法存在问题。

1. 如果一个主题失败，它将终止整个程序。
2. 线程有限。
3. 难以维护代码、调试和支持任何问题。
4. 如果您想从 kafka 中提取特定主题的所有数据，这是不可能的，因为任何配置更改都将适用于所有主题，因此它的操作成本太高。

my_topics = ["topic_1", "topic_2"]

for i in my_topics:
    df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", bootstrap_servers) \
        .option("subscribePattern", i) \
        .load() \
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

    output_df = df \
        .writeStream \
        .format("console") \
        .option("truncate", False) \
        .outputMode("update") \
        .option("checkpointLocation", "s3://<MY_BUCKET>/{}".format(i)) \
        .start()

Answer 2

每个驱动程序节点运行多个 # 并发流当然是合理的。

每个 .start() 在 spark 中都会消耗一定的驱动资源。您的限制因素将是驱动节点上的负载及其可用资源。 100 个以高速连续运行的主题需要分布在多个驱动程序节点上[在 Databricks 中，每个集群有一个驱动程序]。正如您所提到的，Spark 的优势在于多个接收器以及用于转换的统一批处理和流式 API。

另一个问题是处理您可能最终对 S3 进行的小型写入和文件一致性。查看 delta.io 以处理对 S3 的一致且可靠的写入。