Spark job using HDFS storage

Question

I have a long-running Spark Structured Streaming Job running on Google Cloud Dataproc that is using Kafka as both a source and a sink. I am also saving my checkpoints in Google Cloud Storage.

After running for a week, I noticed that it is steadily consuming all of the 100 GB disk storage, saving files to /hadoop/dfs/data/current/BP-315396706-10.128.0.26-1568586969675/current/finalized/....

My understanding is that my Spark job should not have any dependency on local disk storage.

Am I totally misunderstanding this here?

I submitted my job like so:

(cd  app/src/packages/ &&  zip -r mypkg.zip mypkg/ ) && mv app/src/packages/mypkg.zip build
gcloud dataproc jobs submit pyspark \
    --cluster cluster-26aa \
    --region us-central1 \
    --properties ^#^spark.jars.packages=org.apache.spark:spark-streaming-kafka-0-10_2.11:2.4.3,org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 \
    --py-files build/mypkg.zip \
    --max-failures-per-hour 10 \
    --verbosity info \
    app/src/explode_rmq.py

These are the pertinent parts of my job:

Source:

 spark = SparkSession \
        .builder \
        .appName("MyApp") \
        .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    spark.sparkContext.addPyFile('mypkg.zip')

    df = spark \
        .readStream \
        .format("kafka") \
        .options(**config.KAFKA_PARAMS) \
        .option("subscribe", "lsport-rmq-12") \
        .option("startingOffsets", "earliest") \
        .load() \
        .select(f.col('key').cast(t.StringType()), f.col('value').cast(t.StringType()))

Sink:

    sink_kafka_q = sink_df \
        .writeStream \
        .format("kafka") \
        .options(**config.KAFKA_PARAMS) \
        .option("topic", "my_topic") \
        .option("checkpointLocation", "gs://my-bucket-data/checkpoints/my_topic") \
        .start()

user_241899 · Answer

If the memory is not enough, Spark will persist information on the local disk. You can disable the persistence on disk like this:

df.persist(org.apache.spark.storage.StorageLevel.MEMORY_ONLY)

Or you can try to serialize the information to occupy less memory like this

df.persist(org.apache.spark.storage.StorageLevel.MEMORY_ONLY_SER)

Reading the serialized data will be more CPU intensive.

Every dataframe has its distinct serialization level.

For more information: https://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence

user_404264 · Answer

Can you SSH into the master node and run the following command to find out who is consuming the HDFS space?

hdfs df -du -h /

I tested with a simple Spark Pi job,

before running the job:

$ hdfs dfs -du /
34       /hadoop
0        /tmp
2107947  /user

after the job finishes:

$ hdfs dfs -du /user/
0        /user/hbase
0        /user/hdfs
0        /user/hive
0        /user/mapred
0        /user/pig
0        /user/root
2107947  /user/spark
0        /user/yarn
0        /user/zookeeper

$ hdfs dfs -du /user/spark/
2107947  /user/spark/eventlog

Seems it is consumed by Spark eventlog, see spark.eventLog.dir. You can consider compressing eventlog with spark.eventLog.compress=true or disable it with spark.eventLog.enabled=false

Spark job using HDFS storage

2 Answers