File I/O with Snowpark Connect for Spark¶

With Snowpark Connect for Spark, you can read and write data in several file formats using the standard Spark DataFrame reader and writer APIs. You can use internal stages, external stages, and cloud storage locations as sources and destinations. For how to configure stages and cloud URIs, see External data sources with Snowpark Connect for Spark.

When you set Spark-style options (such as header, delimiter, or dateFormat), Snowpark Connect for Spark translates them into the corresponding Snowflake file format options before executing the operation. The tables in the Format-specific options section show exactly how each Spark option maps to its Snowflake counterpart.

Supported formats¶

Reading data¶

Use the standard Spark read API with format short-hands or .format(). Paths can be Snowflake stage notation (for example, @my_stage/path/), cloud URIs configured for your account, or local file paths on the machine where Snowpark Connect for Spark is running. Chain .option() calls to control format behavior.

# Read from a Snowflake stage
df = spark.read.csv("@my_stage/data/")
df = spark.read.json("@my_stage/events/")
df = spark.read.parquet("@my_stage/warehouse/")
df = spark.read.text("@my_stage/logs/")

# Read from a local file
df = spark.read.csv("/tmp/local_data.csv")

# XML supports schema inference, or you can provide an explicit schema
df = spark.read.format("xml").load("@my_stage/exports/")
df = spark.read.format("xml").schema(schema).load("@my_stage/exports/")

# Set format options
df = (
    spark.read.option("header", "true")
    .option("delimiter", ";")
    .option("dateFormat", "yyyy-MM-dd")
    .csv("@my_stage/data/")
)

# Apply an explicit schema
df = spark.read.schema(my_schema).csv("@my_stage/data/")

// Read from a Snowflake stage
Dataset<Row> df = spark.read().csv("@my_stage/data/");
Dataset<Row> df = spark.read().json("@my_stage/events/");
Dataset<Row> df = spark.read().parquet("@my_stage/warehouse/");
Dataset<Row> df = spark.read().text("@my_stage/logs/");

// XML uses .format("xml")
Dataset<Row> df = spark.read().format("xml").load("@my_stage/exports/");
Dataset<Row> df = spark.read().format("xml")
    .option("rowTag", "record")
    .schema(mySchema)
    .load("@my_stage/exports/");

// Set format options
Dataset<Row> df = spark.read()
    .option("header", "true")
    .option("delimiter", ";")
    .option("dateFormat", "yyyy-MM-dd")
    .csv("@my_stage/data/");

// Apply an explicit schema
Dataset<Row> df = spark.read().schema(mySchema).csv("@my_stage/data/");

// Read from a Snowflake stage
val df = spark.read.csv("@my_stage/data/")
val df = spark.read.json("@my_stage/events/")
val df = spark.read.parquet("@my_stage/warehouse/")
val df = spark.read.text("@my_stage/logs/")

// XML uses .format("xml")
val df = spark.read.format("xml").load("@my_stage/exports/")
val df = spark.read.format("xml")
  .option("rowTag", "record")
  .schema(mySchema)
  .load("@my_stage/exports/")

// Set format options
val df = spark.read
  .option("header", "true")
  .option("delimiter", ";")
  .option("dateFormat", "yyyy-MM-dd")
  .csv("@my_stage/data/")

// Apply an explicit schema
val df = spark.read.schema(mySchema).csv("@my_stage/data/")

Reading files with SQL identifiers¶

In addition to the DataFrameReader API, you can read files directly from SQL using Spark’s format-prefix identifier syntax with spark.sql():

spark.sql("SELECT * FROM parquet.`@my_stage/sales/`").show()

spark.sql("SELECT count(*) FROM csv.`@my_stage/raw/orders.csv`").show()

spark.sql("""
  SELECT user_id, event_type
  FROM   json.`@my_stage/events/2025/01/`
  WHERE  event_type IN ('click', 'view')
""").show()

spark.sql("SELECT value FROM text.`@my_stage/logs/access.log`").show()

spark.sql("SELECT * FROM parquet.`@my_stage/sales/`").show();

spark.sql("SELECT count(*) FROM csv.`@my_stage/raw/orders.csv`").show();

spark.sql("SELECT value FROM text.`@my_stage/logs/access.log`").show();

spark.sql("SELECT * FROM parquet.`@my_stage/sales/`").show()

spark.sql("SELECT count(*) FROM csv.`@my_stage/raw/orders.csv`").show()

spark.sql("SELECT value FROM text.`@my_stage/logs/access.log`").show()

Supported prefixes: csv.…, `json.`..., parquet.…, `text.`....

The path inside the backticks is treated the same as it would be by spark.read.<format>(path). Snowflake stage paths (@stage/...), cloud URIs, and local paths all work. Default format options apply (no header for CSV, automatic schema inference, and so on). If you need non-default options, use spark.read instead because the SQL identifier form doesn’t accept per-call options.

You can combine these file reads with the rest of your SQL: joins, aggregations, CTEs, and subqueries all work on top of them.

Writing data¶

Use the DataFrame write API with the same format short-hands.

df.write.csv("@my_stage/out/csv/")
df.write.json("@my_stage/out/json/")
df.write.parquet("@my_stage/out/parquet/")
df.write.text("@my_stage/out/text/")

df.write().csv("@my_stage/out/csv/");
df.write().json("@my_stage/out/json/");
df.write().parquet("@my_stage/out/parquet/");
df.write().text("@my_stage/out/text/");

df.write.csv("@my_stage/out/csv/")
df.write.json("@my_stage/out/json/")
df.write.parquet("@my_stage/out/parquet/")
df.write.text("@my_stage/out/text/")

Reading from and writing to Snowflake tables¶

You can treat Snowflake tables as Spark DataFrame sources and sinks.

# Read
df = spark.read.table("my_database.my_schema.my_table")
df = spark.table("my_table")

# Write
df.write.saveAsTable("my_database.my_schema.my_table")
df.write.mode("overwrite").saveAsTable("my_table")

// Read
Dataset<Row> df = spark.read().table("my_database.my_schema.my_table");
Dataset<Row> df = spark.table("my_table");

// Write
df.write().saveAsTable("my_database.my_schema.my_table");
df.write().mode("overwrite").saveAsTable("my_table");

// Read
val df = spark.read.table("my_database.my_schema.my_table")
val df = spark.table("my_table")

// Write
df.write.saveAsTable("my_database.my_schema.my_table")
df.write.mode("overwrite").saveAsTable("my_table")

Save modes¶

Set mode on the writer to control how writes interact with existing data. Not all modes are available for every format or destination type.

df.write.mode("overwrite").parquet("@my_stage/out/")

df.write().mode("overwrite").parquet("@my_stage/out/");

df.write.mode("overwrite").parquet("@my_stage/out/")

Controlling output file count and size¶

By default, Snowflake decides how to split your output into files. You can control this with standard Spark APIs and Snowpark Connect for Spark-specific options.

coalesce(1) for a single output file¶

The most common Spark idiom for producing one output file is honored:

df.coalesce(1).write.parquet("@my_stage/single_output/")

df.coalesce(1).write().parquet("@my_stage/single_output/");

df.coalesce(1).write.parquet("@my_stage/single_output/")

When coalesce(1) is used without partitionBy, Snowpark Connect for Spark routes the write into a single Snowflake file.

repartition(n) for multiple output files¶

When you request n output files, Snowpark Connect for Spark produces n files with Spark-compatible names:

df.repartition(8).write.parquet("@my_stage/8way_output/")

df.repartition(8).write().parquet("@my_stage/8way_output/");

df.repartition(8).write.parquet("@my_stage/8way_output/")

Rows are distributed across the n target files so each file gets roughly the same amount of data. When you combine repartition(n) with partitionBy, the per-partition directory layout takes precedence and Snowflake controls the file count inside each col=value/ directory.

single and snowflake_max_file_size write options¶

Snowpark Connect for Spark provides two additional write options for explicit control over output layout:

df.write \
    .option("single", "true") \
    .option("snowflake_max_file_size", str(256 * 1024 * 1024)) \
    .parquet("@my_stage/sized_output/")

df.write()
    .option("single", "true")
    .option("snowflake_max_file_size", String.valueOf(256 * 1024 * 1024))
    .parquet("@my_stage/sized_output/");

df.write
  .option("single", "true")
  .option("snowflake_max_file_size", (256 * 1024 * 1024).toString)
  .parquet("@my_stage/sized_output/")

Compression¶

Set compression through .option("compression", "<codec>"). For writes, the default is NONE for CSV, JSON, and text, and SNAPPY for Parquet. For reads, compression is auto-detected if not specified.

For best performance, use splittable compression formats such as BZ2 or SNAPPY. Splittable formats allow Snowflake to decompress and process file chunks in parallel, which is significantly faster for large files than non-splittable formats like GZIP.

Snowpark Connect for Spark normalizes codec names: UNCOMPRESSED becomes NONE, and Spark’s BZIP2 becomes BZ2 for CSV, JSON, and text. Compression can also be inferred from file extensions (.gz, .bz2, .snappy, .deflate).

Parallel file reads¶

Multiple files¶

When you read from a directory or pass a list of file paths, Snowpark Connect for Spark reads the files in parallel automatically. No additional configuration is required.

# Read all files in a directory in parallel
df = spark.read.parquet("@my_stage/data/")

# Read a specific list of files in parallel
df = spark.read.parquet(
    "@my_stage/data/part-001.parquet",
    "@my_stage/data/part-002.parquet",
    "@my_stage/data/part-003.parquet",
)

// Read all files in a directory in parallel
Dataset<Row> df = spark.read().parquet("@my_stage/data/");

// Read a specific list of files in parallel
Dataset<Row> df = spark.read().parquet(
    "@my_stage/data/part-001.parquet",
    "@my_stage/data/part-002.parquet",
    "@my_stage/data/part-003.parquet"
);

// Read all files in a directory in parallel
val df = spark.read.parquet("@my_stage/data/")

// Read a specific list of files in parallel
val df = spark.read.parquet(
  "@my_stage/data/part-001.parquet",
  "@my_stage/data/part-002.parquet",
  "@my_stage/data/part-003.parquet"
)

Single large file¶

Snowpark Connect for Spark can also split a single large file into chunks and read them in parallel, so one large file doesn’t become a bottleneck. This is supported for CSV, JSON, and XML.

CSV: The file must be uncompressed (compression set to none) and multiLine must be false (the default):

df = spark.read.format("csv") \
    .option("compression", "none") \
    .load("@my_stage/large_file.csv")

Dataset<Row> df = spark.read().format("csv")
    .option("compression", "none")
    .load("@my_stage/large_file.csv");

val df = spark.read.format("csv")
  .option("compression", "none")
  .load("@my_stage/large_file.csv")

JSON: The file must be uncompressed (compression set to none) or BZ2-compressed, and multiLine must be false (the default):

df = spark.read.option("compression", "none") \
    .json("@my_stage/large_file.jsonl")

# BZ2-compressed NDJSON also reads in parallel
df = spark.read.option("compression", "bz2") \
    .json("@my_stage/large_file.jsonl.bz2")

Dataset<Row> df = spark.read().json("@my_stage/large_file.jsonl");

// BZ2-compressed NDJSON also reads in parallel
Dataset<Row> df = spark.read().option("compression", "bz2")
    .json("@my_stage/large_file.jsonl.bz2");

val df = spark.read.json("@my_stage/large_file.jsonl")

// BZ2-compressed NDJSON also reads in parallel
val df = spark.read.option("compression", "bz2")
  .json("@my_stage/large_file.jsonl.bz2")

XML: Parallel reads are enabled by default:

df = spark.read.format("xml") \
    .option("rowTag", "record") \
    .load("@my_stage/large_export.xml")

Dataset<Row> df = spark.read().format("xml")
    .option("rowTag", "record")
    .load("@my_stage/large_export.xml");

val df = spark.read.format("xml")
  .option("rowTag", "record")
  .load("@my_stage/large_export.xml")

Automatic exclusion of metadata files¶

When you read from a directory, Snowpark Connect for Spark automatically skips files that Spark also skips, so you don’t have to filter them out manually:

This exclusion applies to CSV, JSON, Parquet, and XML reads at any directory depth, including inside partition subdirectories (for example, year=2025/_SUCCESS is skipped).

Snowpark Connect for Spark also anchors read paths so that files outside your requested prefix are never picked up. For example, reading from @stage/sales/ won’t accidentally include files from @stage/sales_archive/.

If you set pathGlobFilter to a pattern that explicitly matches hidden or metadata files (for example, _*), that pattern takes precedence and those files are included.

Format-specific options¶

Spark options set through .option() are translated into Snowflake file format options before Snowpark Connect for Spark executes the read or write operation. The tables below show how each Spark option maps to its Snowflake counterpart and note any behavioral differences.

Options that don’t have a Snowflake equivalent are ignored. Snowpark Connect for Spark logs a warning for each unsupported option.

CSV options¶

In addition, Snowpark Connect for Spark always sets the following on CSV reads:

• ESCAPE_UNENCLOSED_FIELD = NONE
• ERROR_ON_COLUMN_COUNT_MISMATCH = False
• SKIP_BLANK_LINES = True

JSON options¶

When writing JSON, Snowpark Connect for Spark converts the DataFrame into a single VARIANT column using OBJECT_CONSTRUCT before unloading. On the write path, nullValue is mapped to NULL_IF and compression is mapped to COMPRESSION. Other Spark JSON writer options (such as dateFormat) aren’t applied.

Parquet options¶

Snowpark Connect for Spark also sets these options automatically based on session configuration:

• BINARY_AS_TEXT = False (always)
• USE_LOGICAL_TYPE: controlled by snowpark.connect.parquet.useLogicalType

When writing Parquet, structured complex types (ARRAY, MAP, STRUCT) are cast to VARIANT before unload so that Snowflake’s COPY INTO can produce valid Parquet files.

Text options¶

Text I/O uses a CSV file format internally with FIELD_DELIMITER = NONE.

Text writes always set ESCAPE_UNENCLOSED_FIELD = NONE and FILE_EXTENSION = txt. The DataFrame must contain exactly one string column.

XML options (read only)¶

XML supports schema inference. If no schema is provided, Snowpark Connect for Spark infers the schema from the data by reading the files and merging field types across all input.

Date and timestamp format conversion¶

For CSV reads and writes, Snowpark Connect for Spark automatically converts Java SimpleDateFormat patterns (used by Spark) to Snowflake file format tokens. Common conversions:

Some patterns don’t have exact Snowflake equivalents:

• Unpadded values: Spark’s single-letter patterns (d for day, h for hour) produce unpadded output (for example, 9), but Snowflake always zero-pads (DD produces 09).
• Full day name: Spark’s EEEE (Thursday) maps to abbreviated DY (Thu) in Snowflake.
• Timezone offsets: Spark’s Z (-0500) maps to TZHTZM (-0500) without a colon separator. The colon-separated TZH:TZM format (-05:00) applies to XXX and xxx patterns.

Partitioned data¶

Snowpark Connect for Spark supports Hive-style col=value/ directory layouts for CSV, JSON, and Parquet formats.

Writing partitioned data¶

Use the standard partitionBy(...) API. Snowpark Connect for Spark produces the same col=value/ directory tree that Spark does:

df.write.partitionBy("year", "month").parquet("@my_stage/out/partitioned/")

df.write().partitionBy("year", "month").parquet("@my_stage/out/partitioned/");

df.write.partitionBy("year", "month").parquet("@my_stage/out/partitioned/")

Reading partitioned data¶

Point the reader at the root directory and Snowpark Connect for Spark discovers partition columns from the directory names automatically:

df = spark.read.parquet("@my_stage/out/partitioned/")
df.printSchema()

# Read a specific partition directly by path
df = spark.read.parquet("@my_stage/out/partitioned/year=2025/month=1/")

# Read all partitions and filter on a partition column
df = spark.read.parquet("@my_stage/out/partitioned/")
recent = df.filter(df["year"] == 2025)
recent.show()

Dataset<Row> df = spark.read().parquet("@my_stage/out/partitioned/");
df.printSchema();

// Read a specific partition directly by path
Dataset<Row> jan = spark.read().parquet("@my_stage/out/partitioned/year=2025/month=1/");

// Read all partitions and filter on a partition column
Dataset<Row> all = spark.read().parquet("@my_stage/out/partitioned/");
Dataset<Row> recent = all.filter(col("year").equalTo(2025));
recent.show();

val df = spark.read.parquet("@my_stage/out/partitioned/")
df.printSchema()

// Read a specific partition directly by path
val jan = spark.read.parquet("@my_stage/out/partitioned/year=2025/month=1/")

// Read all partitions and filter on a partition column
val all = spark.read.parquet("@my_stage/out/partitioned/")
val recent = all.filter($"year" === 2025)
recent.show()

Partition discovery behavior:

• Partition columns appear at the end of the schema, in the order they appear in the directory tree.
• Partition value types are inferred from the observed values: all integers become IntegerType, all floating-point values become DoubleType, otherwise StringType. You can override a partition column’s type by supplying a .schema(...) that includes it.
• null values in a partition column are written as the directory segment __HIVE_DEFAULT_PARTITION__ and read back as null.
• Mixed-depth or conflicting layouts (for example, the same key at two different depths) raise an error rather than producing incorrect values.

Dynamic partition overwrite¶

When writing partitioned data to a stage, you can overwrite only the partitions that appear in the current DataFrame instead of deleting all existing data. Set the spark.sql.sources.partitionOverwriteMode session configuration to dynamic:

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df.write.mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet("@my_stage/out/partitioned/")

spark.conf().set("spark.sql.sources.partitionOverwriteMode", "dynamic");

df.write().mode("overwrite")
    .partitionBy("year", "month")
    .parquet("@my_stage/out/partitioned/");

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df.write.mode("overwrite")
  .partitionBy("year", "month")
  .parquet("@my_stage/out/partitioned/")

This removes only the partition subdirectories that match the DataFrame. See Snowpark Connect for Spark properties for more on this configuration property.

Known limitations¶

• Save mode support varies by format: ignore mode is supported only for Parquet file writes and table writes. CSV, JSON, and text file writes raise an error for ignore mode.
• ORC and Avro: These formats aren’t supported for read or write.
• Bucketed writes: bucketBy and sortBy aren’t supported for file or table writes.
• One-sided whitespace trimming: Spark’s ignoreLeadingWhiteSpace and ignoreTrailingWhiteSpace both map to TRIM_SPACE, which trims from both sides. Trimming only leading or only trailing whitespace isn’t possible.
• CSV quote semantics: Spark’s quote option and Snowflake’s FIELD_OPTIONALLY_ENCLOSED_BY have different semantics. Snowflake treats enclosure as optional, while Spark applies mandatory quoting rules.
• Text column constraint: Text writes require exactly one string column in the DataFrame.
• XML MapType: MapType isn’t supported for XML reads. Use StructType to represent key-value structures.
• XML via Spark SQL: Reading XML files with spark.sql() (for example, SELECT * FROM xml.@stage/file.xml``) isn’t supported. Use the DataFrameReader API instead.