KafkaConsumer 对于多线程访问 pyspark 是不安全的

Question

我在 Kafka 中使用 spark 结构化流，但是当我尝试将流写入控制台时出现错误：

Caused by: java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

这是我的代码：

def group_obs(obs_df):
        obs = obs_df.select(f.col("obs.payload.after").alias("obs"))

        filtered_obs_with_value = obs \
            .union(obs.filter("obs.value_datetime is not null")
                   .withColumn("value", f.col("obs.value_datetime"))
                   .withColumn("value_type", f.lit("datetime")))


        grouped_by_obsgroup = filtered_obs_with_value\
                             .groupBy("obs.obs_group_id", "obs.encounter_id")
                             .agg(f.struct(f.col("obs.obs_group_id"),f.collect_list("tempObs").alias("obs")).alias("obs"))

        query = grouped_by_obsgroup \
                .writeStream \
                .outputMode("update") \
                .format("console") \
                .start()

        query.awaitTermination()

raw_obs = kafka_stream.select(from_json(col("value").cast("string"),mySchema)
transformed_obs = group_obs(raw_obs)

Answer 1

您的代码没有什么特别的问题。

这是SPARK-23636 跟踪的一个已知错误。 SPARK-19185 跟踪的直接DStream 也存在类似问题。

根据JIRA票证：

唯一的解决方法是使用 executor-cores = 1 启动我们的应用程序，并启用动态资源分配。

在您的情况下可能会或可能不会接受。