Spark API 옵션 참조

이 페이지에는 데이터를 읽고 쓰는 Spark API에 사용할 수 있는 입력 및 출력 옵션이 나열되어 있습니다.

DataFrameReader 옵션

이러한 옵션은 DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO 및 Auto Loader 사용하여 Azure Databricks 데이터 파일을 읽는 방법을 제어합니다.

Example

다음 예제에서는 JSON 파일을 읽도록 True 설정합니다multiLine.

Python

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

스칼라

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)

일반적인

다음 옵션은 모든 파일 형식에 적용됩니다.

Avro

Csv

Excel

Json

Kafka

Kafka 판독기 옵션의 전체 목록은 DataStreamReader Kafka 옵션을 참조하세요. 다음 옵션은 .를 사용하는 spark.read.format("kafka")일괄 처리 읽기에만 적용합니다.

오크

쪽모이 세공 마루

상태 저장소

이러한 옵션과 테이블 spark.read.format("statestore") 반환 함수를 read_statestore 사용하여 구조적 스트리밍 상태 데이터를 읽습니다. 구조적 스트리밍 상태 정보 읽기를 참조하세요.

텍스트

Xml

DataStreamReader 옵션

이러한 옵션을 DataStreamReader.option() 사용하여 Delta Lake 테이블 및 기타 파일 기반 원본에서 스트리밍 읽기를 구성합니다.

파일 형식 옵션(JSON, CSV, Parquet 등)은 DataFrameReader 옵션을 참조하세요.

자동 로더(cloudFiles.*) 옵션은 자동 로더를 참조하세요.

Example

다음 예제에서는 Delta Lake 테이블 스트림에 대해 설정합니다 maxFilesPerTrigger10 .

Python

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

스칼라

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

일반적인

다음 옵션은 Delta Lake 테이블 및 기타 파일 기반 스트리밍 원본에 적용됩니다.

자동 로더

원본에서 cloudFiles 이러한 옵션을 사용하여 클라우드 스토리지에서 스트리밍 수집을 위한 자동 로더 를 구성합니다. 원본과 cloudFiles 관련된 옵션은 다른 구조적 스트리밍 원본 옵션과 cloudFiles 별도의 네임스페이스에 유지하기 위해 접두사로 사용됩니다.

일반적인

디렉터리 목록

파일 알림

필요한 클라우드 권한, 설정 지침 및 인증 방법을 포함하여 파일 알림 모드를 구성하는 방법에 대한 자세한 내용은 파일 알림 모드에서 자동 로더 스트림 구성을 참조하세요.

(1) 자동 로더는 기본적으로 최상의 노력으로 다음 키-값 태그 쌍을 추가합니다.

• vendor: Databricks
• path: 데이터가 로드되는 위치입니다. 레이블 지정 제한으로 인해 GCP에서 사용할 수 없습니다.
• checkpointLocation: 스트림 검사점의 위치입니다. 레이블 지정 제한으로 인해 GCP에서 사용할 수 없습니다.
• streamId: 스트림의 전역 고유 식별자입니다.

Databricks는 이러한 키 이름을 예약하며 해당 값을 덮어쓸 수 없습니다.

클라우드 관련

자동 로더는 파일 알림 모드에 대한 클라우드 인프라를 구성하는 옵션을 제공합니다. 필요한 클라우드 권한 및 설정 지침은 파일 알림 모드에서 자동 로더 스트림 구성을 참조하세요.

Aws

자동 로더를 선택하여 cloudFiles.useNotifications = true 알림 서비스를 설정하려는 경우에만 다음 옵션을 제공합니다.

cloudFiles.useNotifications = true를 선택하고 자동 로더가 이미 설정한 큐를 사용하도록 하려는 경우에만 다음 옵션을 제공합니다.

AWS 인증 옵션

Databricks 서비스 자격 증명을 사용하려면 다음 인증 옵션을 제공합니다.

Databricks 서비스 자격 증명 또는 IAM 역할을 사용할 수 없는 경우 대신 다음 인증 옵션을 제공할 수 있습니다.

Azure

cloudFiles.useNotifications = true를 지정하고 자동 로더를 통해 알림 서비스를 설정하려는 경우 다음과 같은 모든 옵션의 값을 제공해야 합니다.

Databricks 서비스 자격 증명을 사용할 수 없는 경우 다음 인증 옵션을 대신 제공할 수 있습니다.

자동 로더가 기존 큐를 사용하도록 설정한 cloudFiles.useNotifications = true 경우에만 다음 옵션을 제공합니다.

Gcp

자동 로더는 Databricks 서비스 자격 증명을 활용하여 자동으로 알림 서비스를 설정할 수 있습니다. Databricks 서비스 자격 증명으로 만든 서비스 계정에는 파일 알림 모드에서 자동 로더 스트림 구성에 지정된 권한이 필요합니다.

Databricks 서비스 자격 증명을 사용할 수 없는 경우 Google 서비스 계정을 직접 사용할 수 있습니다. Google 서비스 설정에 따라 서비스 계정을 가정하도록 클러스터를 구성하거나 다음 인증 옵션을 직접 제공할 수 있습니다.

cloudFiles.useNotifications = true를 선택하고 자동 로더가 이미 설정한 큐를 사용하도록 하려는 경우에만 다음 옵션을 제공합니다.

Delta Lake

다음 옵션은 .를 사용하여 spark.readStreamDelta Lake 테이블에서 읽을 때 적용됩니다.

Kafka

다음 옵션 중 하나 spark.readStream.format("kafka") 또는 spark.read.format("kafka")다음을 사용합니다.

다음 옵션은 다음을 사용하여 스트리밍 읽기 spark.readStream.format("kafka")에만 적용합니다.

일괄 처리 읽기 spark.read.format("kafka")에만 적용되는 오프셋 옵션은 DataFrameReader Kafka 옵션을 참조하세요.

Kafka 클라이언트(kafka.*) 및 인증 옵션은 옵션을 참조하세요.

DataFrameWriter 옵션

이러한 옵션을 DataFrameWriter.option() 및 DataFrameWriterV2.option() 사용하여 Azure Databricks 데이터를 쓰는 방법을 제어합니다.

Example

다음 예제에서는 Delta Lake 테이블을 작성하도록 True 설정합니다mergeSchema.

Python

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

스칼라

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

Avro

Delta Lake 및 Apache Iceberg

Csv

Excel

Json

오크

쪽모이 세공 마루

텍스트

Xml

DataStreamWriter 옵션

이러한 옵션을 DataStreamWriter.option() 사용하여 스트리밍 쓰기를 구성합니다.

Example

다음 예제에서는 스트림의 검사점 위치를 설정합니다.

Python

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

스칼라

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

일반적인

콘솔 싱크

Delta Lake

다음 옵션은 을 사용하여 format("delta")Delta Lake 테이블에 스트림을 쓸 때 적용됩니다. 와 같은 overwriteSchemareplaceWhere덮어쓰기 전용 옵션이며 partitionOverwriteMode 스트리밍 쓰기에는 지원되지 않습니다.

파일 싱크

다음 옵션은 파일 기반 형식(Parquet, JSON, CSV, ORC, text)에 스트림을 쓸 때 적용됩니다. 형식별 옵션은 DataFrameWriter 옵션을 참조하세요.

Kafka 싱크

Kafka에 스트림을 쓰기 위한 옵션의 전체 목록은 옵션을 참조하세요.

메모리 싱크

Spark 함수 옵션

일부 Spark SQL 기본 제공 함수는 구문 분석 또는 serialization 동작을 제어하는 맵을 허용 options 합니다. 옵션을 Python dict 또는 Scala Map[String, String] 전달합니다.

Example

다음 예제에서는 형식이 잘못된 레코드를 삭제하는 동안 JSON 열을 구문 분석합니다.

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

스칼라

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

Avro 함수는 해당 DataFrame 옵션과 동일한 옵션을 허용합니다.

• from_avro DataFrameReaderschema_of_avro Avro 옵션을 사용합니다.
• to_avro 에서는 DataFrameWriter Avro 옵션을 사용합니다.

Example

다음 예제에서는 스키마 진화를 사용하도록 설정된 Avro 열을 디코딩합니다.

Python

from pyspark.sql.functions import from_avro

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

스칼라

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

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

또한 스키마 레지스트리의 변형은 from_avro 다음과 같은 옵션을 적용합니다 to_avro .

Csv

CSV 함수는 해당 DataFrame 옵션과 동일한 옵션을 허용합니다.

• from_csv DataFrameReader schema_of_csvCSV 옵션을 사용합니다.
• to_csv 에서는 DataFrameWriter CSV 옵션을 사용합니다.

Example

다음 예제에서는 사용자 지정 구분 기호 및 NULL 값이 있는 CSV를 읽습니다.

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

스칼라

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

JSON 함수는 해당 DataFrame 옵션과 동일한 옵션을 허용합니다.

• from_json DataFrameReaderschema_of_json JSON 옵션을 사용합니다.
• to_json 에서는 DataFrameWriter JSON 옵션을 사용합니다.

Example

다음 예제에서는 필드가 무시되고 서식이 매우 좋은 JSON NULL 을 씁니다.

Python

from pyspark.sql.functions import to_json

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

스칼라

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 파일 to_protobuf 기반 DataSource를 사용하지 않습니다. Protobuf 데이터는 항상 이러한 함수를 사용하여 이진 열로 읽고 작성됩니다. 옵션은 a Map[String, String] 로 전달되며 대/소문자를 구분합니다.

Example

다음 예제에서는 PERMISSIVE 모드를 사용하여 Protobuf 열을 디코딩합니다.

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

스칼라

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

Protobuf 함수는 다음 옵션을 사용합니다.

Xml

XML 함수는 해당 DataFrame 옵션과 동일한 옵션을 허용합니다.

• from_xml DataFrameReaderschema_of_xml XML 옵션을 사용합니다.
• to_xml 에서는 DataFrameWriter XML 옵션을 사용합니다.

Example

다음 예제에서는 사용자 지정 루트 및 행 태그를 사용하여 XML을 씁니다.

Python

from pyspark.sql.functions import to_xml

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

스칼라

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

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