File I/O with Snowpark Connect for Spark¶

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.

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/")

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/")

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¶

# 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"
)

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")

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")

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.

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.

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/")

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()

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/")

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.