Aracılığıyla paylaş

malzeme edilmiş görünüm

Dekoratör @materialized_view, işlem hattında gerçekleştirilmiş görünümleri tanımlamak için kullanılabilir.

Gerçekleştirilmiş görünüm tanımlamak için, veri kaynağında toplu okuma gerçekleştiren bir sorguya uygulayın @materialized_view .

Sözdizimi

from pyspark import pipelines as dp

@dp.materialized_view(
  name="<name>",
  comment="<comment>",
  spark_conf={"<key>" : "<value>", "<key>" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  path="<storage-location-path>",
  partition_cols=["<partition-column>", "<partition-column>"],
  cluster_by_auto = <bool>,
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  schema="schema-definition",
  row_filter = "row-filter-clause",
  private = <bool>)
@dp.expect(...)
def <function-name>():
    return (<query>)

Parametreler

@dp.expect() isteğe bağlı bir beklenti yan tümcesidir. Birden çok beklenti ekleyebilirsiniz. Bkz. Beklentiler.

Parametre Türü Description
fonksiyon function Gerekli. Kullanıcı tanımlı bir sorgudan Apache Spark toplu veri çerçevesi döndüren işlev.
name str Tablo adı. Sağlanmadıysa, varsayılan olarak işlev adını kullanır.
comment str Tablo için bir açıklama.
spark_conf dict Bu sorgunun yürütülmesi için Spark yapılandırmalarının listesi
table_properties dict dict Tablonun tablo özelliklerinden biri.
path str Tablo verileri için bir depolama konumu. Ayarlanmadıysa, tabloyu içeren şema için yönetilen depolama konumunu kullanın.
partition_cols list Tabloyu bölümlendirmek için kullanılacak bir veya daha fazla sütunun listesi.
cluster_by_auto bool Tabloda otomatik sıvı kümelemasını etkinleştirin. Bu, ilk kümeleme anahtarları olarak kullanılacak sütunlarla cluster_by birleştirilebilir ve tanımlanabilir, ardından iş yüküne göre izleme ve otomatik anahtar seçimi güncelleştirmeleri gerçekleştirilir. Bkz. Otomatik sıvı kümeleme.
cluster_by list Tabloda sıvı kümelemeye olanak tanıyın ve kümeleme anahtarları olarak kullanılacak sütunları tanımlayın. Bkz Tablolar için sıvı kümeleme kullanma.
schema str veya StructType Tablo için şema tanımı. Şemalar bir SQL DDL dizesi olarak veya Python StructTypeile tanımlanabilir.
private bool Bir tablo oluşturun, ancak tabloyu meta veri deposunda yayımlamayın. Bu tablo işlem hattına açıktır ancak işlem hattı dışından erişilemez. Özel tablolar işlem hattının ömrü boyunca kalır.
Varsayılan değer: False.
row_filter str (Genel Önizleme) Tablo için satır filtresi yan tümcesi. Bkz. Satır filtreleri ve sütun maskeleriyle tabloları yayımlama.

Şema belirtmek isteğe bağlıdır ve PySpark StructType veya SQL DDL ile yapılabilir. Bir şema belirttiğinizde, isteğe bağlı olarak oluşturulan sütunları, sütun maskelerini ve birincil ve yabancı anahtarları ekleyebilirsiniz. See:

Örnekler

from pyspark import pipelines as dp

# Specify a schema
sales_schema = StructType([
  StructField("customer_id", StringType(), True),
  StructField("customer_name", StringType(), True),
  StructField("number_of_line_items", StringType(), True),
  StructField("order_datetime", StringType(), True),
  StructField("order_number", LongType(), True)]
)
@dp.materialized_view(
  comment="Raw data on sales",
  schema=sales_schema)
def sales():
  return ("...")

# Specify a schema with SQL DDL, use a generated column, and set clustering columns
@dp.materialized_view(
  comment="Raw data on sales",
  schema="""
    customer_id STRING,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime))
    """,
  cluster_by = ["order_day_of_week", "customer_id"])
def sales():
  return ("...")

# Specify partition columns
@dp.materialized_view(
  comment="Raw data on sales",
  schema="""
    customer_id STRING,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime))
    """,
  partition_cols = ["order_day_of_week"])
def sales():
  return ("...")

# Specify table constraints
@dp.materialized_view(
   schema="""
    customer_id STRING NOT NULL PRIMARY KEY,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime)),
    CONSTRAINT fk_customer_id FOREIGN KEY (customer_id) REFERENCES main.default.customers(customer_id)
    """)
def sales():
   return ("...")

# Specify a row filter and column mask
@dp.materialized_view(
   schema="""
    id int COMMENT 'This is the customer ID',
    name string COMMENT 'This is the customer full name',
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn USING COLUMNS (region)
    """,
  row_filter = "ROW FILTER catalog.schema.us_filter_fn ON (region, name)")
def sales():
   return ("...")
  • Last updated on