Spark-Streaming 最早挂起 kafka 起始偏移量（Kafka 2，spark 2.4.3）
Spark-Streaming hangs with kafka starting offset at earliest (Kafka 2, spark 2.4.3)
我在使用 Spark-Streaming 和 Kafka 时遇到问题。虽然 运行 一个从 Kafka 主题消费并将微批处理结果输出到终端的示例程序，但当我设置选项时我的工作似乎挂起：
df.option("startingOffsets", "earliest")
从最新的偏移量开始作业工作正常，结果会在每个微批流通过时打印到终端。
我在想这可能是一个资源问题——我正在尝试阅读一个包含大量数据的主题。但是我似乎没有 memory/cpu 问题（运行 这个工作与本地 [*] 集群）。这项工作似乎从未真正开始，只是挂在线上：
19/09/17 15:21:37 INFO Metadata: Cluster ID: JFXVL24JQ3K4CEbE-VA58A
val sc = new SparkConf().setMaster("local[*]").setAppName("spark-test")
val streamContext = new StreamingContext(sc, Seconds(1))
val spark = SparkSession.builder().appName("spark-test")
.getOrCreate()
val topic = "topic.with.alotta.data"
//subscribe tokafka
val df = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", topic)
.option("startingOffsets", "earliest")
.load()
//write
df.writeStream
.outputMode("append")
.format("console")
.option("truncate", "false")
.start()
.awaitTermination()
我希望看到打印到控制台的结果....但是，应用程序似乎就像我提到的那样挂起。有什么想法吗？感觉像是一个 spark 资源问题（因为我是 运行 本地人 "cluster" 针对具有大量数据的主题。我是否遗漏了有关流式数据帧的性质？
写入控制台会导致所有数据在每次触发时都收集到驱动程序的内存中。由于您目前没有限制批次的大小，这意味着整个主题内容都在驱动程序中累积。参见 https://spark.apache.org/docs/2.4.3/structured-streaming-programming-guide.html#output-sinks
设置批量大小限制应该可以解决您的问题。
从 Kafka 读取时尝试添加 maxOffsetsPerTrigger 设置...
val df = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", topic)
.option("startingOffsets", "earliest")
.option("maxOffsetsPerTrigger", 1000)
.load()
详情见https://spark.apache.org/docs/2.4.3/structured-streaming-kafka-integration.html。
我在使用 Spark-Streaming 和 Kafka 时遇到问题。虽然 运行 一个从 Kafka 主题消费并将微批处理结果输出到终端的示例程序，但当我设置选项时我的工作似乎挂起：
df.option("startingOffsets", "earliest")
从最新的偏移量开始作业工作正常，结果会在每个微批流通过时打印到终端。
我在想这可能是一个资源问题——我正在尝试阅读一个包含大量数据的主题。但是我似乎没有 memory/cpu 问题（运行 这个工作与本地 [*] 集群）。这项工作似乎从未真正开始，只是挂在线上：
19/09/17 15:21:37 INFO Metadata: Cluster ID: JFXVL24JQ3K4CEbE-VA58A
val sc = new SparkConf().setMaster("local[*]").setAppName("spark-test")
val streamContext = new StreamingContext(sc, Seconds(1))
val spark = SparkSession.builder().appName("spark-test")
.getOrCreate()
val topic = "topic.with.alotta.data"
//subscribe tokafka
val df = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", topic)
.option("startingOffsets", "earliest")
.load()
//write
df.writeStream
.outputMode("append")
.format("console")
.option("truncate", "false")
.start()
.awaitTermination()
我希望看到打印到控制台的结果....但是，应用程序似乎就像我提到的那样挂起。有什么想法吗？感觉像是一个 spark 资源问题（因为我是 运行 本地人 "cluster" 针对具有大量数据的主题。我是否遗漏了有关流式数据帧的性质？
写入控制台会导致所有数据在每次触发时都收集到驱动程序的内存中。由于您目前没有限制批次的大小，这意味着整个主题内容都在驱动程序中累积。参见 https://spark.apache.org/docs/2.4.3/structured-streaming-programming-guide.html#output-sinks
设置批量大小限制应该可以解决您的问题。
从 Kafka 读取时尝试添加 maxOffsetsPerTrigger 设置...
val df = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", topic)
.option("startingOffsets", "earliest")
.option("maxOffsetsPerTrigger", 1000)
.load()
详情见https://spark.apache.org/docs/2.4.3/structured-streaming-kafka-integration.html。