结构化流错误 py4j.protocol.Py4JNetworkError:Java 端的答案为空

Posted

技术标签:

【中文标题】结构化流错误 py4j.protocol.Py4JNetworkError:Java 端的答案为空【英文标题】:Structured Streaming error py4j.protocol.Py4JNetworkError: Answer from Java side is empty 【发布时间】:2018-10-14 17:38:26 【问题描述】:

我正在尝试使用 PySpark 和结构化流 (Spark 2.3) 在两个 Kafka 流之间进行左外连接。

import os
import time

from pyspark.sql.types import *
from pyspark.sql.functions import from_json, col, struct, explode, get_json_object
from ast import literal_eval
from pyspark.sql import SparkSession
from pyspark.sql.functions import expr

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell'

spark = SparkSession \
    .builder \
    .appName("Spark Kafka Structured Streaming") \
    .getOrCreate()

schema_impressions = StructType() \
    .add("id_req", StringType()) \
    .add("ts_imp_request", TimestampType()) \
    .add("country", StringType()) \
    .add("TS_IMPRESSION", TimestampType()) 

schema_requests = StructType() \
    .add("id_req", StringType()) \
    .add("page", StringType()) \
    .add("conntype", StringType()) \
    .add("TS_REQUEST", TimestampType()) 

impressions = spark.readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "ip-ec2.internal:9092") \
  .option("subscribe", "ssp.datascience_impressions") \
  .load()

requests = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "ip-ec2.internal:9092") \
  .option("subscribe", "ssp.datascience_requests") \
  .option("startingOffsets", "latest") \
  .load()

query_requests = requests \
        .select(col("timestamp"), col("key").cast("string"), from_json(col("value").cast("string"), schema_requests).alias("parsed")) \
        .select(col("timestamp").alias("timestamp_req"), "parsed.id_req", "parsed.page", "parsed.conntype", "parsed.TS_REQUEST") \
        .withWatermark("timestamp_req", "120 seconds") 

query_impressions = impressions \
        .select(col("timestamp"), col("key").cast("string"), from_json(col("value").cast("string"), schema_impressions).alias("parsed")) \
        .select(col("timestamp").alias("timestamp_imp"), col("parsed.id_req").alias("id_imp"), "parsed.ts_imp_request", "parsed.country", "parsed.TS_IMPRESSION") \
        .withWatermark("timestamp_imp", "120 seconds") 

query_requests.printSchema()        
query_impressions.printSchema()

> root  
|-- timestamp_req: timestamp (nullable = true)  
|-- id_req: string (nullable = true)  
|-- page: string (nullable = true)  
|-- conntype: string (nullable = true)  
|-- TS_REQUEST: timestamp (nullable = true)
> 
> root  |-- timestamp_imp: timestamp (nullable = true)  
|-- id_imp: string (nullable = true)  
|-- ts_imp_request: timestamp (nullable = true)  
|-- country: string (nullable = true)  
|-- TS_IMPRESSION: timestamp (nullable = true)

在简历中,我将从两个 Kafka Streams 中获取数据,在接下来的几行中,我将尝试使用 ID 进行连接。

rawQuery = query_requests.join(query_impressions,  expr(""" 
    (id_req = id_imp AND 
    timestamp_imp >= timestamp_req AND 
    timestamp_imp <= timestamp_req + interval 5 minutes) 
    """), 
  "leftOuter")

rawQuery = rawQuery \
        .writeStream \
        .format("parquet") \
        .option("checkpointLocation", "/home/jovyan/streaming/applicationHistory") \
        .option("path", "/home/jovyan/streaming").start()
print(rawQuery.status)

'message': '处理新数据', 'isDataAvailable': True, 'isTriggerActive':True 错误:root:发送命令时出现异常。 Traceback(最近一次通话最后一次):文件 “/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 1062,在 send_command 中 raise Py4JNetworkError("Answer from Java side is empty") py4j.protocol.Py4JNetworkError: Answer from Java side is empty

在处理上述异常的过程中,又发生了一个异常:

Traceback(最近一次调用最后一次):文件 “/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 908,在 send_command 中 response = connection.send_command(command) 文件“/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 1067,在 send_command 中 “接收时出错”,e,proto.ERROR_ON_RECEIVE)py4j.protocol.Py4JNetworkError:接收时出错 错误:py4j.java_gateway:尝试连接时发生错误 Java 服务器 (127.0.0.1:33968) Traceback(最近一次调用最后一次): 文件 "/opt/conda/lib/python3.6/site-packages/IPython/core/interactiveshell.py", 第 2910 行,在 run_code 中 exec(code_obj, self.user_global_ns, self.user_ns) 文件“”,第 3 行,在 打印(rawQuery.status)文件“/opt/conda/lib/python3.6/site-packages/pyspark/sql/streaming.py”, 第 114 行,处于状态 返回 json.loads(self._jsq.status().json()) 文件“/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 1160,在通话中 答案,self.gateway_client,self.target_id,self.name)文件“/opt/conda/lib/python3.6/site-packages/pyspark/sql/utils.py”,行 63,在装饰 返回 f(*a, **kw) 文件“/opt/conda/lib/python3.6/site-packages/py4j/protocol.py”,第 328 行, 在 get_return_value format(target_id, ".", name)) py4j.protocol.Py4JError: 调用o92.status时出错

在处理上述异常的过程中,又发生了一个异常:

Traceback(最近一次调用最后一次):文件 "/opt/conda/lib/python3.6/site-packages/IPython/core/interactiveshell.py", 第 1828 行,在 showtraceback 中 stb = value._render_traceback_() AttributeError: 'Py4JError' 对象没有属性 '_render_traceback_'

在处理上述异常的过程中,又发生了一个异常:

Traceback(最近一次调用最后一次):文件 “/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 第852章 connection = self.deque.pop() IndexError: pop from an empty deque

我正在使用 Jupyter Notebook 在本地运行 Spark。在 spark/conf/spark-defaults.conf 我有:

# Example:
# spark.master                     spark://master:7077
# spark.eventLog.enabled           true
# spark.eventLog.dir               hdfs://namenode:8021/directory
# spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.driver.memory             15g
# spark.executor.extraJavaOptions  -XX:+PrintGCDetails -Dkey=value -Dnumbers="one two three"

如果我在上一个错误之后尝试使用 Spark,我会收到该错误:

错误:root:发送命令时出现异常。回溯(最近 最后调用):文件 “/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 1062,在 send_command 中 raise Py4JNetworkError("Answer from Java side is empty") py4j.protocol.Py4JNetworkError: Answer from Java side is empty

在处理上述异常的过程中,又发生了一个异常:

Traceback(最近一次调用最后一次):文件 “/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 908,在 send_command 中 response = connection.send_command(command) 文件“/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py”,行 1067,在 send_command 中 "接收时出错", e, proto.ERROR_ON_RECEIVE) py4j.protocol.Py4JNetworkError: 接收时出错

【问题讨论】:

【参考方案1】:

我解决了这个问题!基本上,由于某种原因,该问题与 Jupyter Notebook 有关。我删除了之前代码的下一行:

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell'

我使用控制台运行代码:

> spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 spark_structured.py

这样,我可以毫无问题地运行所有代码。

如果你有同样的问题,你也可以改变 spark-default.conf 并增加 spark.driver.memoryspark.executor.memory

【讨论】:

以上是关于结构化流错误 py4j.protocol.Py4JNetworkError:Java 端的答案为空的主要内容,如果未能解决你的问题,请参考以下文章

结构化流错误 py4j.protocol.Py4JNetworkError:Java 端的答案为空

使用 Boost 发送序列化结构但得到输入流错误

Spark结构化流内存绑定

使用 Pyspark 从结构化流数据帧构建 Spark ML 管道模型

卡夫卡结构化流 java.lang.NoClassDefFoundError

pyspark结构化流kafka - py4j.protocol.Py4JJavaError:调用o41.save时发生错误