RPC response exceeds maximum data length

[jcasse]

Problem

When executing a PySpark application form within a Docker container set up for Ozone as per
https://ozone.apache.org/docs/1.3.0/recipe/sparkozonefsk8s.html and https://ozone.apache.org/docs/1.3.0/interface/ofs.html,
get exception org.apache.hadoop.ipc.RpcException: RPC response exceeds maximum data length (reproduced in step 11 below).

Note that there is no error when executing the same application form within the base Spark image (reproduced in step 6 below).
Also, in step 10 below, a different PySpark application appears to write/read to/from Ozone without issue.

---

Steps to Reproduce

1. Make temp directory

         mkdir ~/tmp
         cd ~/tmp

2. Create PySpark minimal streaming example
   https://spark.apache.org/docs/latest/api/python/reference/pyspark.ss/api/pyspark.sql.streaming.DataStreamReader.load.html
   Create streaming-minimal.py:

         from pyspark.sql import SparkSession
         import tempfile
         import time

         spark = (
             SparkSession
             .builder
             .appName("StructuredNetworkWordCount")
             .getOrCreate()
             )

         with tempfile.TemporaryDirectory() as d:

             # Write a temporary JSON file to read it
             (
              spark
              .createDataFrame([(100, "Hyukjin Kwon"),], ["age", "name"])
              .write.mode("overwrite")
              .format("json")
              .save(d)
              )

              # Start a streaming query to read the JSON file
              query = (
                  spark
                  .readStream
                  .schema("age INT, name STRING")
                  .format("json")
                  .load(d)
                  .writeStream
                  .format("console")
                  .start()
                  )

              time.sleep(3)

              query.stop()

3. Verify it runs (on local machine)
   Set up Python virtual environment:

         python3.10 -m venv venv
         source venv/bin/activate
         pip install pyspark==3.5.0

Run PySpark streaming example:

python streaming-minimal.py

Output should look something like:

         23/12/29 16:19:50 WARN Utils: Your hostname, my-host resolves to a loopback address: 127.0.1.1; using 192.168.1.110 instead (on interface enp114s0)
         23/12/29 16:19:50 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
         Setting default log level to "WARN".
         To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
         23/12/29 16:19:50 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
         23/12/29 16:19:52 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-cbd45dea-e049-4379-b727-97a0590e55fc. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.
         23/12/29 16:19:52 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
         -------------------------------------------
         Batch: 0
         -------------------------------------------
         +---+------------+
         |age|        name|
         +---+------------+
         |100|Hyukjin Kwon|
         +---+------------+

4. Build Spark image using Spark's script
   https://spark.apache.org/downloads.html
   Download Apache Spark 3.5.0 pre-built for Apache Hadoop 3.3 and later:
   https://www.apache.org/dyn/closer.lua/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz
   Untar Spark download.
   https://spark.apache.org/docs/latest/running-on-kubernetes.html#docker-images
   From spark-3.5.0-bin-hadoop3/:

./bin/docker-image-tool.sh -r my-repo -t my-tag -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile build

5. Start Docker container
   From ~/tmp/:

         docker run -it --rm \
                --mount type=bind,source="$(pwd)",target=/opt/spark/mount \
                my-repo/spark-py:my-tag \
                bash

6. Verify PySpark streaming example still runs from within Docker container

         ../bin/spark-submit \
             --master local \
             ../mount/streaming-minimal.py

7. Deploy Ozone cluster with Docker Compose
   https://ozone.apache.org/docs/1.3.0/start/startfromdockerhub.html
   This will create the Docker network =ozone_default=, which we will use when running our custom image below.

8. Customize PySpark Docker image to work with Ozone
   https://ozone.apache.org/docs/1.3.0/interface/ofs.html
   https://ozone.apache.org/docs/1.3.0/recipe/sparkozonefsk8s.html
   Create the following files in ~/tmp/:
   Dockerfile:

         FROM my-repo/spark-py:my-tag

         # Used by Ivy package manager (when using --package option in spark-submit)
         USER root
         RUN mkdir /opt/spark/.ivy2

         ADD ozone/core-site.xml /opt/hadoop/conf/core-site.xml
         ADD ozone/ozone-site.xml /opt/hadoop/conf/ozone-site.xml
         ENV HADOOP_CONF_DIR=/opt/hadoop/conf
         ENV SPARK_EXTRA_CLASSPATH=/opt/hadoop/conf
         ADD ozone/ozone-filesystem-hadoop3.jar /opt/ozone-filesystem-hadoop3.jar

core-site.xml:

         <configuration>
           <property>
             <name>fs.ofs.impl</name>
             <value>org.apache.hadoop.fs.ozone.RootedOzoneFileSystem</value>
           </property>
           <property>
             <name>fs.defaultFS</name>
             <value>ofs://om:9874/</value>
           </property>
         </configuration>

ozone-site.xml:

         <configuration>

         <property><name>ozone.om.address</name><value>om-0.om</value></property>
         <property><name>ozone.scm.client.address</name><value>scm-0.scm</value></property>
         <property><name>ozone.scm.block.client.address</name><value>scm-0.scm</value></property>
         <property><name>ozone.scm.names</name><value>scm-0.scm</value></property>

         <property><name>hdds.scm.safemode.min.datanode</name><value>3</value></property>
         <property><name>ozone.datanode.pipeline.limit</name><value>1</value></property>

         <property><name>ozone.scm.datanode.id.dir</name><value>/data</value></property>
         <property><name>ozone.metadata.dirs</name><value>/data/metadata</value></property>
         <property><name>hdds.datanode.dir</name><value>/data/storage</value></property>

         </configuration>

ozone-filesystem-hadoop3.jar:

docker cp ozone-om-1:/opt/hadoop/share/ozone/lib/ozone-filesystem-hadoop3-1.3.0.jar ozone-filesystem-hadoop3.jar

Build custom image:

docker build -t my-repo/my-tag-ozone .

9. Start Docker container (set up for Ozone) in =ozone_network=

         docker run -it --rm \
                --network ozone_default \
                --mount type=bind,source="$(pwd)",target=/opt/spark/mount \
                my-repo/spark-py:my-tag-ozone \
                bash

10. Verify PySpark application can write/read to/from Ozone
    Create the following file:
    iceberg-table.py:

         from pyspark.sql import SparkSession

         spark = (
             SparkSession
             .builder

             .appName("IcebergTable")

             # Iceberg
             # https://spark.apache.org/docs/latest/configuration.html
             .config("spark.jars.packages",
                     "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.2"
                     ",org.apache.ozone:ozone-filesystem-hadoop3-client:1.3.0"
                     ",org.apache.ozone:ozone-filesystem-hadoop3:1.3.0")
             .config("spark.sql.extensions",
                     "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")

             # Iceberg Catalog "mycatalog"
             # https://www.dremio.com/blog/3-ways-to-use-python-with-apache-iceberg/
             .config("spark.sql.catalog.mycatalog",
                     "org.apache.iceberg.spark.SparkCatalog")
             .config("spark.sql.catalog.mycatalog.type",
                     "hadoop")
             .config("spark.sql.catalog.mycatalog.warehouse",
                     "ofs://om/warehouse")
             .getOrCreate())

         print('- Spark configuration '.ljust(80, '-'))
         confs: list = spark.sparkContext.getConf().getAll()
         for conf in confs:
             print(conf)
         print('-' * 80)

         df = spark.sql("CREATE TABLE IF NOT EXISTS mycatalog.mytable (id bigint, data string) USING iceberg")
         df.show()

         print('- Schema of Iceberg table '.ljust(80, '-'))
         df.printSchema()

         df = spark.sql("INSERT INTO mycatalog.mytable VALUES (1, 'a'), (2, 'b'), (3, 'c')")
         df.show()

         df = spark.sql("SELECT * FROM mycatalog.mytable")
         df.show()

         df = spark.sql("DROP TABLE IF EXISTS mycatalog.mytable PURGE")

         spark.stop()

Run PySpark application from within Docker container:

         ../bin/spark-submit \
             --master local \
             --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.2,org.apache.ozone:ozone-filesystem-hadoop3-client:1.3.0 \
             ../mount/iceberg-table.py

11. Execute PySpark application (that works in simple Spark container) inside custom container set up for Ozone
    The following throws exception: org.apache.hadoop.ipc.RpcException: RPC response exceeds maximum data length:

         ../bin/spark-submit \
             --master local \
             --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.2,org.apache.ozone:ozone-filesystem-hadoop3-client:1.3.0 \
             ../mount/streaming-minimal.py

[jcasse]

The Ozone Manger port 9874 in core-site.xml is wrong:

        <configuration>
           <property>
             <name>fs.ofs.impl</name>
             <value>org.apache.hadoop.fs.ozone.RootedOzoneFileSystem</value>
           </property>
           <property>
             <name>fs.defaultFS</name>
             <value>ofs://om:9874/</value>
           </property>
         </configuration>

After changing it to 9862, the PySpark application in step 11 worked.