/
githubmirror
/
spark
Обзор
Документация
Войти
/
githubmirror
/
spark
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
examples/src/main/python/sql/datasource.py
499 строк
17 KB
Calvin Qin
[SPARK-57354][SQL] Add ignoredPathSegmentRegex data source option and config
23 июн 2026, 23:28
23 июн 2026, 23:28
0417f91
Код
Авторство
О чём код?
# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """ A simple example demonstrating Spark SQL data sources. Run with: ./bin/spark-submit examples/src/main/python/sql/datasource.py """ from pyspark.sql import SparkSession # $example on:schema_merging$ from pyspark.sql import Row # $example off:schema_merging$ def generic_file_source_options_example(spark: SparkSession) -> None: # $example on:ignore_corrupt_files$ # enable ignore corrupt files via the data source option # dir1/file3.json is corrupt from parquet's view test_corrupt_df0 = spark.read.option("ignoreCorruptFiles", "true")\ .parquet("examples/src/main/resources/dir1/", "examples/src/main/resources/dir1/dir2/") test_corrupt_df0.show() # +-------------+ # | file| # +-------------+ # |file1.parquet| # |file2.parquet| # +-------------+ # enable ignore corrupt files via the configuration spark.sql("set spark.sql.files.ignoreCorruptFiles=true") # dir1/file3.json is corrupt from parquet's view test_corrupt_df1 = spark.read.parquet("examples/src/main/resources/dir1/", "examples/src/main/resources/dir1/dir2/") test_corrupt_df1.show() # +-------------+ # | file| # +-------------+ # |file1.parquet| # |file2.parquet| # +-------------+ # $example off:ignore_corrupt_files$ # $example on:recursive_file_lookup$ recursive_loaded_df = spark.read.format("parquet")\ .option("recursiveFileLookup", "true")\ .load("examples/src/main/resources/dir1") recursive_loaded_df.show() # +-------------+ # | file| # +-------------+ # |file1.parquet| # |file2.parquet| # +-------------+ # $example off:recursive_file_lookup$ # $example on:ignored_path_segment_regex$ # An empty regex surfaces files that are hidden by default (e.g. names starting with "_" or ".") surfaced_df = spark.read.format("parquet")\ .option("ignoredPathSegmentRegex", "")\ .load("examples/src/main/resources/dir1") surfaced_df.show() # $example off:ignored_path_segment_regex$ spark.sql("set spark.sql.files.ignoreCorruptFiles=false") # $example on:load_with_path_glob_filter$ df = spark.read.load("examples/src/main/resources/dir1", format="parquet", pathGlobFilter="*.parquet") df.show() # +-------------+ # | file| # +-------------+ # |file1.parquet| # +-------------+ # $example off:load_with_path_glob_filter$ # $example on:load_with_modified_time_filter$ # Only load files modified before 07/1/2050 @ 08:30:00 df = spark.read.load("examples/src/main/resources/dir1", format="parquet", modifiedBefore="2050-07-01T08:30:00") df.show() # +-------------+ # | file| # +-------------+ # |file1.parquet| # +-------------+ # Only load files modified after 06/01/2050 @ 08:30:00 df = spark.read.load("examples/src/main/resources/dir1", format="parquet", modifiedAfter="2050-06-01T08:30:00") df.show() # +-------------+ # | file| # +-------------+ # +-------------+ # $example off:load_with_modified_time_filter$ def basic_datasource_example(spark: SparkSession) -> None: # $example on:generic_load_save_functions$ users_df = spark.read.load("examples/src/main/resources/users.parquet") users_df.select("name", "favorite_color").write.save("namesAndFavColors.parquet") # $example off:generic_load_save_functions$ # $example on:write_partitioning$ users_df = spark.read.load("examples/src/main/resources/users.parquet") users_df.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet") # $example off:write_partitioning$ # $example on:write_partition_and_bucket$ users_df = spark.read.parquet("examples/src/main/resources/users.parquet") (users_df.write .partitionBy("favorite_color") .bucketBy(42, "name") .saveAsTable("users_partitioned_bucketed")) # $example off:write_partition_and_bucket$ # $example on:manual_load_options$ people_df = spark.read.load("examples/src/main/resources/people.json", format="json") people_df.select("name", "age").write.save("namesAndAges.parquet", format="parquet") # $example off:manual_load_options$ # $example on:manual_load_options_csv$ people_df = spark.read.load( "examples/src/main/resources/people.csv", format="csv", sep=";", inferSchema="true", header="true" ) # $example off:manual_load_options_csv$ # $example on:manual_save_options_orc$ users_df = spark.read.orc("examples/src/main/resources/users.orc") (users_df.write.format("orc") .option("orc.bloom.filter.columns", "favorite_color") .option("orc.dictionary.key.threshold", "1.0") .option("orc.column.encoding.direct", "name") .save("users_with_options.orc")) # $example off:manual_save_options_orc$ # $example on:manual_save_options_parquet$ users_df = spark.read.parquet("examples/src/main/resources/users.parquet") (users_df.write.format("parquet") .option("parquet.bloom.filter.enabled#favorite_color", "true") .option("parquet.bloom.filter.expected.ndv#favorite_color", "1000000") .option("parquet.enable.dictionary", "true") .option("parquet.page.write-checksum.enabled", "false") .save("users_with_options.parquet")) # $example off:manual_save_options_parquet$ # $example on:write_sorting_and_bucketing$ people_df = spark.read.json("examples/src/main/resources/people.json") people_df.write.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed") # $example off:write_sorting_and_bucketing$ # $example on:direct_sql$ df = spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`") # $example off:direct_sql$ spark.sql("DROP TABLE IF EXISTS people_bucketed") spark.sql("DROP TABLE IF EXISTS users_partitioned_bucketed") def parquet_example(spark: SparkSession) -> None: # $example on:basic_parquet_example$ peopleDF = spark.read.json("examples/src/main/resources/people.json") # DataFrames can be saved as Parquet files, maintaining the schema information. peopleDF.write.parquet("people.parquet") # Read in the Parquet file created above. # Parquet files are self-describing so the schema is preserved. # The result of loading a parquet file is also a DataFrame. parquetFile = spark.read.parquet("people.parquet") # Parquet files can also be used to create a temporary view and then used in SQL statements. parquetFile.createOrReplaceTempView("parquetFile") teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19") teenagers.show() # +------+ # | name| # +------+ # |Justin| # +------+ # $example off:basic_parquet_example$ def parquet_schema_merging_example(spark: SparkSession) -> None: # $example on:schema_merging$ # spark is from the previous example. # Create a simple DataFrame, stored into a partition directory sc = spark.sparkContext squaresDF = spark.createDataFrame(sc.parallelize(range(1, 6)) .map(lambda i: Row(single=i, double=i ** 2))) squaresDF.write.parquet("data/test_table/key=1") # Create another DataFrame in a new partition directory, # adding a new column and dropping an existing column cubesDF = spark.createDataFrame(sc.parallelize(range(6, 11)) .map(lambda i: Row(single=i, triple=i ** 3))) cubesDF.write.parquet("data/test_table/key=2") # Read the partitioned table mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table") mergedDF.printSchema() # The final schema consists of all 3 columns in the Parquet files together # with the partitioning column appeared in the partition directory paths. # root # |-- double: long (nullable = true) # |-- single: long (nullable = true) # |-- triple: long (nullable = true) # |-- key: integer (nullable = true) # $example off:schema_merging$ def json_dataset_example(spark: SparkSession) -> None: # $example on:json_dataset$ # spark is from the previous example. sc = spark.sparkContext # A JSON dataset is pointed to by path. # The path can be either a single text file or a directory storing text files path = "examples/src/main/resources/people.json" peopleDF = spark.read.json(path) # The inferred schema can be visualized using the printSchema() method peopleDF.printSchema() # root # |-- age: long (nullable = true) # |-- name: string (nullable = true) # Creates a temporary view using the DataFrame peopleDF.createOrReplaceTempView("people") # SQL statements can be run by using the sql methods provided by spark teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19") teenagerNamesDF.show() # +------+ # | name| # +------+ # |Justin| # +------+ # Alternatively, a DataFrame can be created for a JSON dataset represented by # an RDD[String] storing one JSON object per string jsonStrings = ['{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}'] otherPeopleRDD = sc.parallelize(jsonStrings) otherPeople = spark.read.json(otherPeopleRDD) otherPeople.show() # +---------------+----+ # | address|name| # +---------------+----+ # |[Columbus,Ohio]| Yin| # +---------------+----+ # $example off:json_dataset$ def csv_dataset_example(spark: SparkSession) -> None: # $example on:csv_dataset$ # spark is from the previous example sc = spark.sparkContext # A CSV dataset is pointed to by path. # The path can be either a single CSV file or a directory of CSV files path = "examples/src/main/resources/people.csv" df = spark.read.csv(path) df.show() # +------------------+ # | _c0| # +------------------+ # | name;age;job| # |Jorge;30;Developer| # | Bob;32;Developer| # +------------------+ # Read a csv with delimiter, the default delimiter is "," df2 = spark.read.option("delimiter", ";").csv(path) df2.show() # +-----+---+---------+ # | _c0|_c1| _c2| # +-----+---+---------+ # | name|age| job| # |Jorge| 30|Developer| # | Bob| 32|Developer| # +-----+---+---------+ # Read a csv with delimiter and a header df3 = spark.read.option("delimiter", ";").option("header", True).csv(path) df3.show() # +-----+---+---------+ # | name|age| job| # +-----+---+---------+ # |Jorge| 30|Developer| # | Bob| 32|Developer| # +-----+---+---------+ # You can also use options() to use multiple options df4 = spark.read.options(delimiter=";", header=True).csv(path) # "output" is a folder which contains multiple csv files and a _SUCCESS file. df3.write.csv("output") # Read all files in a folder, please make sure only CSV files should present in the folder. folderPath = "examples/src/main/resources" df5 = spark.read.csv(folderPath) df5.show() # Wrong schema because non-CSV files are read # +-----------+ # | _c0| # +-----------+ # |238val_238| # | 86val_86| # |311val_311| # | 27val_27| # |165val_165| # +-----------+ # $example off:csv_dataset$ def text_dataset_example(spark: SparkSession) -> None: # $example on:text_dataset$ # spark is from the previous example sc = spark.sparkContext # A text dataset is pointed to by path. # The path can be either a single text file or a directory of text files path = "examples/src/main/resources/people.txt" df1 = spark.read.text(path) df1.show() # +-----------+ # | value| # +-----------+ # |Michael, 29| # | Andy, 30| # | Justin, 19| # +-----------+ # You can use 'lineSep' option to define the line separator. # The line separator handles all `\r`, `\r\n` and `\n` by default. df2 = spark.read.text(path, lineSep=",") df2.show() # +-----------+ # | value| # +-----------+ # | Michael| # | 29\nAndy| # | 30\nJustin| # | 19\n| # +-----------+ # You can also use 'wholetext' option to read each input file as a single row. df3 = spark.read.text(path, wholetext=True) df3.show() # +--------------------+ # | value| # +--------------------+ # |Michael, 29\nAndy...| # +--------------------+ # "output" is a folder which contains multiple text files and a _SUCCESS file. df1.write.csv("output") # You can specify the compression format using the 'compression' option. df1.write.text("output_compressed", compression="gzip") # $example off:text_dataset$ def jdbc_dataset_example(spark: SparkSession) -> None: # $example on:jdbc_dataset$ # Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods # Loading data from a JDBC source jdbcDF = spark.read \ .format("jdbc") \ .option("url", "jdbc:postgresql:dbserver") \ .option("dbtable", "schema.tablename") \ .option("user", "username") \ .option("password", "password") \ .load() jdbcDF2 = spark.read \ .jdbc("jdbc:postgresql:dbserver", "schema.tablename", properties={"user": "username", "password": "password"}) # Specifying dataframe column data types on read jdbcDF3 = spark.read \ .format("jdbc") \ .option("url", "jdbc:postgresql:dbserver") \ .option("dbtable", "schema.tablename") \ .option("user", "username") \ .option("password", "password") \ .option("customSchema", "id DECIMAL(38, 0), name STRING") \ .load() # Saving data to a JDBC source jdbcDF.write \ .format("jdbc") \ .option("url", "jdbc:postgresql:dbserver") \ .option("dbtable", "schema.tablename") \ .option("user", "username") \ .option("password", "password") \ .save() jdbcDF2.write \ .jdbc("jdbc:postgresql:dbserver", "schema.tablename", properties={"user": "username", "password": "password"}) # Specifying create table column data types on write jdbcDF.write \ .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") \ .jdbc("jdbc:postgresql:dbserver", "schema.tablename", properties={"user": "username", "password": "password"}) # $example off:jdbc_dataset$ def xml_dataset_example(spark: SparkSession) -> None: # $example on:xml_dataset$ # Primitive types (Int, String, etc) and Product types (case classes) encoders are # supported by importing this when creating a Dataset. # An XML dataset is pointed to by path. # The path can be either a single xml file or more xml files path = "examples/src/main/resources/people.xml" peopleDF = spark.read.option("rowTag", "person").format("xml").load(path) # The inferred schema can be visualized using the printSchema() method peopleDF.printSchema() # root # |-- age: long (nullable = true) # |-- name: string (nullable = true) # Creates a temporary view using the DataFrame peopleDF.createOrReplaceTempView("people") # SQL statements can be run by using the sql methods provided by spark teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19") teenagerNamesDF.show() # +------+ # | name| # +------+ # |Justin| # +------+ # Alternatively, a DataFrame can be created for an XML dataset represented by a Dataset[String] xmlStrings = [""" <person> <name>laglangyue</name> <job>Developer</job> <age>28</age> </person> """] xmlRDD = spark.sparkContext.parallelize(xmlStrings) otherPeople = spark.read \ .option("rowTag", "person") \ .xml(xmlRDD) otherPeople.show() # +---+---------+----------+ # |age| job| name| # +---+---------+----------+ # | 28|Developer|laglangyue| # +---+---------+----------+ # $example off:xml_dataset$ if __name__ == "__main__": spark = SparkSession \ .builder \ .appName("Python Spark SQL data source example") \ .getOrCreate() basic_datasource_example(spark) generic_file_source_options_example(spark) parquet_example(spark) parquet_schema_merging_example(spark) json_dataset_example(spark) csv_dataset_example(spark) text_dataset_example(spark) jdbc_dataset_example(spark) xml_dataset_example(spark) spark.stop()