[jira] [Created] (SPARK-48492) batch-read parquet files written by streaming returns non-nullable fields in schema

["Julien Peloton (Jira)" <ji...@apache.org>]

Julien Peloton created SPARK-48492:
--------------------------------------

             Summary: batch-read parquet files written by streaming returns non-nullable fields in schema
                 Key: SPARK-48492
                 URL: https://issues.apache.org/jira/browse/SPARK-48492
             Project: Spark
          Issue Type: New Feature
          Components: Structured Streaming
    Affects Versions: 3.4.1
         Environment: python --version
Python 3.9.13

 

spark-submit --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.4.1
      /_/
                        
Using Scala version 2.12.17, OpenJDK 64-Bit Server VM, 1.8.0_302
Branch HEAD
Compiled by user centos on 2023-06-19T23:01:01Z
Revision 6b1ff22dde1ead51cbf370be6e48a802daae58b6
            Reporter: Julien Peloton


Hello,

In the documentation, it is stated that

> When reading Parquet files, all columns are automatically converted to be nullable for compatibility reasons.

While this seems correct for batch DataFrames, I have a counter example for streaming ones:

```python

from pyspark.sql import SparkSession
from pyspark.sql import Row
import pyspark.sql.functions as F

spark = SparkSession.builder.getOrCreate()
spark.sparkContext.setLogLevel("WARN")

df = spark.createDataFrame(
    [
        Row(a=1, b=2.0, c="toto"),
        Row(a=3, b=4.0, c="titi"),
        Row(a=10, b=11.0, c="tutu"),
    ]
)

df = df.withColumn('d', F.lit(1.0))

print("Original dataframe")
df.printSchema()

# Write this on disk
df.write.parquet('static.parquet')

# Now load a stream
df_stream = (
    spark.readStream.format("parquet")
    .schema(df.schema)
    .option("path", "static.parquet")
    .option("latestFirst", False)
    .load()
)

df_stream = df_stream.withColumn('e', F.lit("error"))

print("Streaming dataframe")
df_stream.printSchema()

 

# Now write the dataframe using writestream
query = (
    df_stream.writeStream.outputMode("append")
    .format("parquet")
    .option("checkpointLocation", 'test_parquet_checkpoint')
    .option("path", 'test_parquet')
    .trigger(availableNow=True)
    .start()
)

spark.streams.awaitAnyTermination()

# Now read back
df_stream_2 = spark.read.format("parquet").load("test_parquet")
print("Static dataframe from the streaming job")
df_stream_2.printSchema()

```

which outputs:

```

Original dataframe
root
 |-- a: long (nullable = true)
 |-- b: double (nullable = true)
 |-- c: string (nullable = true)
 |-- d: double (nullable = false)

Streaming dataframe
root
 |-- a: long (nullable = true)
 |-- b: double (nullable = true)
 |-- c: string (nullable = true)
 |-- d: double (nullable = true)
 |-- e: string (nullable = false)

Static dataframe from the streaming job
root
 |-- a: long (nullable = true)
 |-- b: double (nullable = true)
 |-- c: string (nullable = true)
 |-- d: double (nullable = true)
 |-- e: string (nullable = false)

```

So the column `d` is correctly set to `nullable = true`, but not the column `e`, which stays non-nullable. Is that expected? According to this [old PR]([https://github.com/apache/spark/pull/25382)] it was supposed to be resolved. Any ideas?



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org