create_table

Important

這項功能位於 測試版 (Beta) 中。

在管線中使用該 create_table() 函式建立一個受管理的資料表，由一個或多個 append_flow 聲明撰寫。 將通話與一位或多位@append_flow(target=...)裝飾者配對create_table()，這些設計師會寫入表格。 多個流程可以針對同一個受管理的資料表。

關於 SQL 的對應，請參見 CREATE TABLE ...流暢。

Syntax

from pyspark import pipelines as dp

dp.create_table(
  name = "<table-name>",
  comment = "<comment>",
  spark_conf={"<key>" : "<value>", "<key>" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  partition_cols=["<partition-column>", "<partition-column>"],
  path="<storage-location-path>",
  schema="schema-definition",
  expect_all = {"<key>" : "<value>", "<key>" : "<value>"},
  expect_all_or_drop = {"<key>" : "<value>", "<key>" : "<value>"},
  expect_all_or_fail = {"<key>" : "<value>", "<key>" : "<value>"},
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  cluster_by_auto = False,
  row_filter = "row-filter-clause",
  private = False
)

Parameters

Parameter 類型 Description
name str Required. 資料表名稱。
comment str 數據表的描述。
spark_conf dict 這是執行此查詢時的 Spark 配置清單。
table_properties dict 一個用於資料表的dict資料表屬性。
partition_cols list 用於分割數據表的一或多個欄列表。
path str 數據表數據的儲存位置。 如果未設定，請使用包含數據表之架構的受控儲存位置。
schema str 或 StructType 數據表的架構定義。 架構可以定義為 SQL DDL 字串，或使用 Python StructType。
expect_all、expect_all_or_drop、expect_all_or_fail dict 表格的數據質量限制。 提供相同的行為，並使用與預期裝飾函式相同的語法，但實作為參數。 請參閱 預期。
cluster_by list 在數據表上啟用液體叢集，並定義要當做叢集索引鍵使用的數據行。 請參閱 針對數據表使用液體叢集。
cluster_by_auto bool 在桌子上啟用自動液體聚集。 可以結合 來 cluster_by 定義初始的群集鍵。 請參閱 自動液體群集。
row_filter str （公開預覽）數據表的數據列篩選子句。 請參閱 使用資料列篩選和欄位遮罩發佈資料表，。
private bool 當 True時，會建立一個私有資料表，該表不會發佈到目錄，且只能在管線內存取。 預設為 False。

Limitations

  • 受管理資料表不支援變更資料擷取（CDC）變更流程。 create_auto_cdc_flow() 或 create_auto_cdc_from_snapshot_flow() 是鎖定受管理資料表失敗。 CDC 目標使用 create_streaming_table（）
  • 僅 append_flow支援管理資料表。 不支援替換流replace_flow / FLOW ... REPLACE WHERE（）。
  • 管理資料表僅支援 Unity Catalog 的管線。
  • 你不能在管理資料表中重複使用現有串流資料表的名稱。

Example

from pyspark import pipelines as dp

dp.create_table("combined")

@dp.append_flow(target="combined")
def from_a():
    return spark.readStream.table("source_a")

@dp.append_flow(target="combined")
def from_b():
    return spark.readStream.table("source_b")
  • Last updated on