Informations de référence sur les options d’API Spark

Cette page répertorie les options d’entrée et de sortie disponibles pour les API Spark qui lisent et écrivent des données.

Options DataFrameReader

Utilisez ces options avec DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO et Auto Loader pour contrôler la façon dont Azure Databricks lit les fichiers de données.

Example

L’exemple suivant définit multiLine la valeur pour True lire des fichiers JSON :

Python

df = spark.read.format("json").option("multiLine", True).load("/path/to/data")

Scala

val df = spark.read.format("json").option("multiLine", "true").load("/path/to/data")

SQL

SELECT * FROM read_files("/path/to/data", format => "json", multiLine => true)

Commun

Les options suivantes s’appliquent à tous les formats de fichier.

Avro

CSV

Excel

JSON

Kafka

Pour obtenir la liste complète des options de lecteur Kafka, consultez les options Kafka DataStreamReader. Les options suivantes s’appliquent uniquement aux lectures par lots à l’aide spark.read.format("kafka")de .

ORC

Parquet

Magasin d’état

Utilisez ces options avec spark.read.format("statestore") ou la read_statestore fonction table pour lire les données d’état structured Streaming. Consultez Lire les informations d’état de Structured Streaming.

Texte

XML

Options DataStreamReader

Utilisez ces options pour DataStreamReader.option() configurer les lectures de diffusion en continu à partir de tables Delta Lake et d’autres sources basées sur des fichiers.

Pour connaître les options de format de fichier (JSON, CSV, Parquet et autres), consultez les options DataFrameReader.

Pour connaître les options du chargeur automatique (cloudFiles.*), consultez Le chargeur automatique.

Example

L’exemple suivant définit maxFilesPerTrigger la valeur pour 10 un flux de table Delta Lake :

Python

df = spark.readStream.format("delta").option("maxFilesPerTrigger", 10).load("/path/to/delta-table")

Scala

val df = spark.readStream.format("delta").option("maxFilesPerTrigger", "10").load("/path/to/delta-table")

Commun

Les options suivantes s’appliquent aux tables Delta Lake et à d’autres sources de diffusion en continu basées sur des fichiers.

Chargeur automatique

Utilisez ces options avec la cloudFiles source pour configurer le chargeur automatique pour l’ingestion de streaming à partir du stockage cloud. Les options spécifiques à la cloudFiles source sont préfixées cloudFiles pour les conserver dans un espace de noms distinct des autres options de source Structured Streaming .

Commun

Liste d’annuaires

Notification de fichier

Pour plus d’informations sur la configuration du mode de notification de fichier, notamment les autorisations cloud requises, les instructions d’installation et les méthodes d’authentification, consultez Configurer des flux de chargeur automatique en mode de notification de fichier.

(1) Auto Loader ajoute les paires de balises clé-valeur suivantes par défaut dans la mesure du possible :

• vendor: Databricks
• path : emplacement à partir duquel les données sont chargées. Non disponible dans GCP en raison de limitations des libellés.
• checkpointLocation: emplacement du point de contrôle du flux. Non disponible dans GCP en raison de limitations des libellés.
• streamId : Identificateur unique au niveau mondial pour le flux.

Databricks réserve ces noms de clés et vous ne pouvez pas remplacer leurs valeurs.

Spécifique au cloud

Le chargeur automatique fournit des options pour configurer l’infrastructure cloud pour le mode de notification de fichier. Pour obtenir les autorisations cloud requises et les instructions de configuration, consultez Configurer des flux de chargeur automatique en mode de notification de fichier.

AWS

Fournissez les options suivantes uniquement si vous choisissez cloudFiles.useNotifications = true et que vous souhaitez que le chargeur automatique configure les services de notification pour vous :

Fournissez l’option suivante uniquement si vous choisissez cloudFiles.useNotifications = true et voulez qu’Auto Loader utilise une file d’attente que vous avez déjà configurée :

Options d’authentification AWS

Fournissez l’option d’authentification suivante pour utiliser les informations d’identification d’un service Databricks :

Lorsque les informations d’identification du service Databricks ou les rôles IAM ne sont pas disponibles, vous pouvez fournir les options d’authentification suivantes à la place :

Azure

Vous devez fournir des valeurs pour toutes les options suivantes si vous spécifiez cloudFiles.useNotifications = true et souhaitez qu’Auto Loader configure les services de notification pour vous :

Si les informations d’identification d’un service Databricks ne sont pas disponibles, vous pouvez fournir les options d’authentification suivantes à la place :

Indiquez l’option suivante uniquement si vous définissez cloudFiles.useNotifications = true et que vous souhaitez que le chargeur automatique utilise une file d’attente existante :

GCP

L'Auto Loader peut configurer automatiquement les services de notification pour vous en utilisant les identifiants de service Databricks . Le compte de service créé avec les informations d’identification du service Databricks nécessite les autorisations spécifiées dans Configurer les flux du chargeur automatique en mode de notification de fichier.

Si les informations d’identification d’un service Databricks ne sont pas disponibles, vous pouvez utiliser directement des comptes de service Google. Vous pouvez configurer votre cluster pour qu’il assume un compte de service en suivant configuration du service Google ou en fournissant directement les options d’authentification suivantes :

Fournissez l’option suivante uniquement si vous choisissez cloudFiles.useNotifications = true et voulez qu’Auto Loader utilise une file d’attente que vous avez déjà configurée :

Delta Lake

Les options suivantes s’appliquent lors de la lecture à partir d’une table Delta Lake à l’aide spark.readStreamde .

Kafka

Utilisez ces options avec l’une ou l’autre spark.readStream.format("kafka") des options spark.read.format("kafka")suivantes :

Les options suivantes s’appliquent uniquement aux lectures de diffusion en continu avec spark.readStream.format("kafka"):

Pour les options de décalage qui s’appliquent uniquement aux lectures par lots avec spark.read.format("kafka"), consultez les options Kafka DataFrameReader.

Pour connaître les options de client (kafka.*) et d’authentification Kafka, consultez Options.

Options DataFrameWriter

Utilisez ces options avec DataFrameWriter.option() et DataFrameWriterV2.option() pour contrôler la façon dont Azure Databricks écrit des données.

Example

L’exemple suivant définit mergeSchema la valeur pour True écrire une table Delta Lake :

Python

df.write.format("delta").option("mergeSchema", True).saveAsTable("my_table")

Scala

df.write.format("delta").option("mergeSchema", "true").saveAsTable("my_table")

Avro

Delta Lake et Apache Iceberg

CSV

Excel

JSON

ORC

Parquet

Texte

XML

Options DataStreamWriter

Utilisez ces options DataStreamWriter.option() pour configurer les écritures de diffusion en continu.

Example

L’exemple suivant définit l’emplacement de point de contrôle d’un flux :

Python

(df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start("/path/to/table"))

Scala

df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start("/path/to/table")

Commun

Récepteur de console

Delta Lake

Les options suivantes s’appliquent lors de l’écriture d’un flux dans une table Delta Lake à l’aide format("delta")de . Les options de remplacement uniquement telles que overwriteSchema, replaceWhereet partitionOverwriteMode ne sont pas prises en charge pour les écritures de streaming.

Récepteur de fichiers

L’option suivante s’applique lors de l’écriture d’un flux dans des formats basés sur des fichiers (Parquet, JSON, CSV, ORC, texte). Pour obtenir des options spécifiques au format, consultez les options DataFrameWriter.

Récepteur Kafka

Pour obtenir la liste complète des options d’écriture de flux dans Kafka, consultez Options.

Récepteur de mémoire

Options de fonction Spark

Certaines fonctions intégrées Spark SQL acceptent une carte qui contrôle le comportement d’analyse options ou de sérialisation. Passez des options en tant que Python dict ou scala Map[String, String].

Example

L’exemple suivant analyse une colonne JSON lors de la suppression d’enregistrements mal formés :

Python

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([StructField("name", StringType())])
df = df.withColumn("parsed", from_json("json_col", schema, {"mode": "DROPMALFORMED"}))

Scala

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types._

val schema = StructType(Seq(StructField("name", StringType)))
val df = df.withColumn("parsed", from_json(col("json_col"), schema, Map("mode" -> "DROPMALFORMED")))

Avro

Les fonctions Avro acceptent les mêmes options que les options de DataFrame correspondantes :

• from_avro et schema_of_avro utilisez les options DataFrameReader Avro.
• to_avro utilise les options DataFrameWriter Avro.

Example

L’exemple suivant décode une colonne Avro avec l’évolution du schéma activée :

Python

from pyspark.sql.functions import from_avro

df = df.withColumn("decoded", from_avro("avro_col", json_schema, {"avroSchemaEvolutionMode": "restart"}))

Scala

import org.apache.spark.sql.avro.functions.from_avro

val df = df.withColumn("decoded", from_avro(col("avro_col"), jsonSchema, Map("avroSchemaEvolutionMode" -> "restart")))

En outre, les variantes du Registre de schémas des from_avroto_avro options suivantes sont les suivantes :

CSV

Les fonctions CSV acceptent les mêmes options que les options dataFrame correspondantes :

• from_csv et schema_of_csv utiliser les options CSV DataFrameReader.
• to_csv utilise les options CSV DataFrameWriter.

Example

L’exemple suivant lit csv avec un séparateur et NULL une valeur personnalisés :

Python

from pyspark.sql.functions import from_csv
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

schema = StructType([StructField("id", IntegerType()), StructField("name", StringType())])
df = df.withColumn("parsed", from_csv("csv_col", schema, {"sep": "|", "nullValue": "N/A"}))

Scala

import org.apache.spark.sql.functions.from_csv
import org.apache.spark.sql.types._

val schema = StructType(Seq(StructField("id", IntegerType), StructField("name", StringType)))
val df = df.withColumn("parsed", from_csv(col("csv_col"), schema, Map("sep" -> "|", "nullValue" -> "N/A")))

JSON

Les fonctions JSON acceptent les mêmes options que les options de DataFrame correspondantes :

• from_json et schema_of_json utilisez les options JSON DataFrameReader.
• to_json utilise les options JSON DataFrameWriter.

Example

L’exemple suivant écrit JSON avec NULL des champs ignorés et une mise en forme assez activée :

Python

from pyspark.sql.functions import to_json

df = df.withColumn("json_str", to_json("struct_col", {"pretty": "true", "ignoreNullFields": "true"}))

Scala

import org.apache.spark.sql.functions.to_json

val df = df.withColumn("json_str", to_json(col("struct_col"), Map("pretty" -> "true", "ignoreNullFields" -> "true")))

Protobuf

from_protobuf et to_protobuf n’utilisez pas de Source de données basée sur des fichiers. Les données Protobuf sont toujours lues et écrites sous forme de colonnes binaires à l’aide de ces fonctions. Les options sont passées en tant que Map[String, String] respectant la casse.

Example

L’exemple suivant décode une colonne Protobuf à l’aide du mode PERMISSIVE :

Python

from pyspark.sql.functions import from_protobuf

df = df.withColumn("decoded", from_protobuf("proto_col", "MyMessage", "/path/to/descriptor.desc",
    {"mode": "PERMISSIVE", "enums.as.ints": "true"}))

Scala

import org.apache.spark.sql.protobuf.functions.from_protobuf

val df = df.withColumn("decoded", from_protobuf(col("proto_col"), "MyMessage", "/path/to/descriptor.desc",
    Map("mode" -> "PERMISSIVE", "enums.as.ints" -> "true")))

Les fonctions Protobuf utilisent les options suivantes :

XML

Les fonctions XML acceptent les mêmes options que les options de DataFrame correspondantes :

• from_xml et schema_of_xml utilisez les options XML DataFrameReader.
• to_xml utilise les options XML DataFrameWriter.

Example

L’exemple suivant écrit du code XML avec des balises racines et de lignes personnalisées :

Python

from pyspark.sql.functions import to_xml

df = df.withColumn("xml_str", to_xml("struct_col", {"rootTag": "records", "rowTag": "record"}))

Scala

import org.apache.spark.sql.functions.to_xml

val df = df.withColumn("xml_str", to_xml(col("struct_col"), Map("rootTag" -> "records", "rowTag" -> "record")))