きっとカンタンなことなんんだろうなと思っていても、少しでも疑問に思ったら確かめるだけのことだ。saveAsTable と save、どう違うの？と。試してわかったことはテーブルのタイプである。

MANAGED table
Lakehouse テーブルのソースファイルは、CREATE TABLEで自動的に用意され、DELETE TABLEでソースファイルも削除される。
EXTERNAL table
CREATE TABLEでも Lakehouse テーブルのソースファイルは自動的に用意されない。DELETE TABLEでもソースファイルは削除されない。

pyspark.sql.DataFrameWriter.saveAsTable

MANAGED table

PySpark : Runtime 1.2

(spark.range(5)
    .write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("managed_table"))

Spark SQL : Runtime 1.2

CREATE OR REPLACE TABLE managed_table
USING DELTA
AS SELECT * FROM range(5)

commitInfo : Runtime 1.2

{
    "commitInfo": {
        "timestamp": 0000000000000,
        "operation": "CREATE OR REPLACE TABLE AS SELECT",
        "operationParameters": {
            "isManaged": "true",
            "description": null,
            "partitionBy": "[]",
            "properties": "{}"
        },
        "isolationLevel": "Serializable",
        "isBlindAppend": false,
        "operationMetrics": {
            "numFiles": "1",
            "numOutputRows": "5",
            "numOutputBytes": "854"
        },
        "tags": {
            "VORDER": "true"
        },
        "engineInfo": "Apache-Spark/3.4.1.5.3-117503204 Delta-Lake/2.4.0.9",
        "txnId": "********-****-****-****-************"
    }
}

EXTERNAL table

PySpark : Runtime 1.2

(spark.range(5)
    .write
    .format("delta")
    .mode("overwrite")
    .saveAsTable(
        "external_table"
        , path="Files/external_table"))

Spark SQL : Runtime 1.2

CREATE OR REPLACE TABLE external_table
USING DELTA
LOCATION "Files/external_table"
AS SELECT * FROM range(5)

PySpark : Runtime 1.2

df_metadata = spark.sql("DESCRIBE EXTENDED table_name")
(df_metadata.filter(
    (df_metadata.col_name == "Name")
    | (df_metadata.col_name == "Type")
    | (df_metadata.col_name == "Provider")
).show(truncate=False))

col_name	data_type	comment
Name	spark_catalog.`lakehouse_name`.`table_name`
Type	MANAGED, EXTERNAL
Provider	delta

Spark SQL : Runtime 1.2

DROP TABLE IF EXISTS external_table;

CREATE EXTERNAL TABLE external_table
LOCATION "Files/contents";

pyspark.sql.DataFrameWriter.save

Save contents

PySpark : Runtime 1.2

(spark.range(5)
    .write
    .format("delta")
    .mode("overwrite")
    .save("Files/contents"))

結果的に MANAGED table

結果的には MANAGED table になる。commitInfo は異なる。

PySpark : Runtime 1.2

(spark.range(5)
    .write
    .format("delta")
    .mode("overwrite")
    .save("Tables/managed_table"))

commitInfo : Runtime 1.2

{
    "commitInfo": {
        "timestamp": 0000000000000,
        "operation": "WRITE",
        "operationParameters": {
            "mode": "Overwrite",
            "partitionBy": "[]"
        },
        "isolationLevel": "Serializable",
        "isBlindAppend": false,
        "operationMetrics": {
            "numFiles": "1",
            "numOutputRows": "5",
            "numOutputBytes": "854"
        },
        "tags": {
            "VORDER": "true"
        },
        "engineInfo": "Apache-Spark/3.4.1.5.3-117503204 Delta-Lake/2.4.0.9",
        "txnId": "********-****-****-****-************"
    }
}

その他