I am trying to load a large (larger than my system memory) json from s3 bucket to spark Dataframe. but getting below error.
org.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 138345560; received: 67924532
- I am using spark 2.4.7 and Hadoop 2.7 distribution, with python 3.7.
- I have copied "hadoop-aws:2.7.3.jar" & "aws-sdk:1.7.4.jar" in
- Sparkhome/jars/ directory. Using JDK 1.8 (set SPARK_HOME, JAVA_HOME, HADOOP_HOME) environment variables already.
- previously had python 3.8, java 15, spark 3.0 with hadoop 2.7 - downgraded because of this error.
# pyspark_job.py
from pyspark.sql import SparkSession
def create_spark_session():
"""
Create spark session.
Returns:
spark (SparkSession) - spark session
"""
spark = SparkSession \
.builder \
.master("local[4]") \
.appName("Myappname") \
.getOrCreate()
return spark
def setConfigs(spark):
"""
setting the environment configuration properties for s3 access
arguments:
spark (SparkSession) - spark session
"""
ak = 'accesskey' # as of now I am using the keys directly, once i get the basic functionality working, the keys will be taken from property file/env variable
sk = 'secretkey'
sc = spark.sparkContext
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", ak)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", sk)
sc._jsc.hadoopConfiguration().set("fs.s3a.readahead.range", '512M')
def process_events_data(spark, input_path):
"""Process the event data from s3 bucket as json files and get specific info.
Arguments:
spark (SparkSession) - spark session connected to local cluster
input_path (str) - AWS S3 bucket path for source data
returns
Data Frame with event data
"""
df = spark.read.format('json').load(input_path)
#df = spark.read.json("input_path")
return df
def main():
bn = 'bucketname'
sf = 'prefix/'
fp = bn+sf
input_path = 's3a://bucketname/prefix/filename.json'
spark = create_spark_session()
setConfigs(spark)
edf = process_events_data(spark, input_path)
if __name__ == '__main__':
main()
Error trace:
20/09/29 22:32:35 WARN TaskSetManager: Lost task 1.0 in stage 0.0 (TID 1, localhost, executor driver): TaskKilled (Stage cancelled) Traceback (most recent call last): File "pyspark_job.py", line 61, in main() File "pyspark_job.py", line 58, in main process_events_data(spark, input_path) File "pyspark_job.py", line 41, in process_events_data df = spark.read.format('json').load(input_path) File "spark_dir\spark-2.4.7_h2.7\python\lib\pyspark.zip\pyspark\sql\readwriter.py", line 166, in load File "spark_dir\spark-2.4.7_h2.7\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py", line 1257, in call File "spark_dir\spark-2.4.7_h2.7\python\lib\pyspark.zip\pyspark\sql\utils.py", line 63, in deco File "spark_dir\spark-2.4.7_h2.7\python\lib\py4j-0.10.7-src.zip\py4j\protocol.py", line 328, in get_return_value py4j.protocol.Py4JJavaError: An error occurred while calling o42.load. : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost, executor driver): org.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 138345560; received: 71284276
already tried below solutions but still getting the error: Premature end of Content-Length delimited message body SparkException while reading from S3 using Pyspark
I am not sure if some configuration is missing? or something wrong on S3 side? or do i need to read the json input line by line? This is my first spark program, can anyone help.