파일 메타데이터 열

_metadata 열을 사용하여 입력 파일에 대한 메타데이터 정보를 가져올 수 있습니다. _metadata 열은 숨겨진 열이며 모든 입력 파일 형식에 사용할 수 있습니다. 반환된 _metadata DataFrame에 열을 포함하려면 원본을 지정하는 읽기 쿼리에서 명시적으로 선택해야 합니다.

데이터 원본에 _metadata열이 포함된 경우 쿼리는 파일 메타데이터가 아닌 데이터 원본에서 열을 반환합니다.

지원되는 메타데이터

_metadata 열은 다음 필드를 포함하고 있는 STRUCT.

예시

기본 파일 기반 데이터 원본 판독기에서 사용

파이썬

df = spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("*", "_metadata")

display(df)

'''
Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f0.csv",                |
| Debbie  | 18  |    "file_name": "f0.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
'''

스칼라

val df = spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("*", "_metadata")

display(df_population)

/* Result:
+---------+-----+----------------------------------------------------+
|   name  | age |                 _metadata                          |
+=========+=====+====================================================+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f0.csv",                |
| Debbie  | 18  |    "file_name": "f0.csv",                          |
|         |     |    "file_size": 12,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-07-02 01:05:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
|         |     | {                                                  |
|         |     |    "file_path": "dbfs:/tmp/f1.csv",                |
| Frank   | 24  |    "file_name": "f1.csv",                          |
|         |     |    "file_size": 10,                                |
|         |     |    "file_block_start": 0,                          |
|         |     |    "file_block_length": 12,                        |
|         |     |    "file_modification_time": "2021-12-20 02:06:21" |
|         |     | }                                                  |
+---------+-----+----------------------------------------------------+
*/

특정 필드 선택

파이썬

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("_metadata.file_name", "_metadata.file_size")

스칼라

spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("_metadata.file_name", "_metadata.file_size")

필터에서 사용

파이썬

spark.read \
  .format("csv") \
  .schema(schema) \
  .load("dbfs:/tmp/*") \
  .select("*") \
  .filter(col("_metadata.file_name") == lit("test.csv"))

스칼라

spark.read
  .format("csv")
  .schema(schema)
  .load("dbfs:/tmp/*")
  .select("*")
  .filter(col("_metadata.file_name") === lit("test.csv"))

COPY INTO (레거시)에서 사용

COPY INTO my_delta_table
FROM (
  SELECT *, _metadata FROM 'abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData'
)
FILEFORMAT = CSV

자동 로더에서 사용

원본 데이터에 이름이 지정된 열이 포함된 경우 이름을 _metadata.로 바꿉니다 source_metadata. 이름을 바꾸지 않으면 대상 테이블의 파일 메타데이터 열에 액세스할 수 없습니다. 쿼리는 대신 원본 열을 반환합니다.

파이썬

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .schema(schema) \
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData") \
  .selectExpr("*", "_metadata as source_metadata") \
  .writeStream \
  .option("checkpointLocation", checkpointLocation) \
  .start(targetTable)

스칼라

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .schema(schema)
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData")
  .selectExpr("*", "_metadata as source_metadata")
  .writeStream
  .option("checkpointLocation", checkpointLocation)
  .start(targetTable)

foreachBatch를 사용하고 스트리밍 데이터 프레임에 파일 메타데이터 열을 포함하려는 경우, foreachBatch 함수 호출 전에 스트리밍 읽기 DataFrame에서 이를 참조해야 합니다. 함수 내의 파일 메타데이터 열만 참조하는 foreachBatch 경우 열은 포함되지 않습니다.

파이썬

spark.readStream \
  .format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData") \
  .select("*", "metadata") \
  .writeStream \
  .foreachBatch(...)

스칼라

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .load("abfss://my-container-name@storage-account-name.dfs.core.windows.net/csvData")
  .select("*", "metadata")
  .writeStream
  .foreachBatch(...)

관련된 문서

• 자동 로더
• 구조적 스트리밍
• COPY INTO(레거시)