CREATE TABLE ... FLOW (işlem hatları)

Important

Bu özellik Beta sürümündedir.

CREATE TABLE ... FLOW bir veya daha fazla akış tarafından yazılmış bir işlem hattında yönetilen tablo oluşturmak için deyimini kullanın.

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

Birden çok kaynağı tek bir yönetilen tabloda gruplandırmak için CREATE FLOW (işlem hatları) ile onu hedefleyen birkaç akış bildirin:

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    Oluşturulacak yönetilen tablonun adı. Ad uygun değilse, tablo işlem hattının hedef şemasında oluşturulur. Ad zaten bir akış tablosuna ait olmamalıdır.

  • tablo_özellikleri

    İsteğe bağlı olarak sütunları, bunların türlerini, özelliklerini ve açıklamalarını tanımlar. Atlanırsa, şema akış sorgusundan çıkarılır.

  • CONSTRAINT expectation_name EXPECT (expectation_expr) [ İHLAL DURUMUNDA { FAIL UPDATE | SATIRI SİL } ]

    Yönetilen tabloya veri kalitesi beklentileri ekler. Bu veri kalitesi beklentileri zamanla takip edilebilir ve boru hattının olay günlüğü üzerinden erişilebilir. FAIL UPDATE bir beklenti, hem tabloyu oluştururken hem de tabloyu yenilerken işlemenin başarısız olmasına neden olur. Beklenti karşılanmadığında, DROP ROW satırın tamamının silinmesine neden olur. Bkz. İşlem hattı beklentileriyle veri kalitesini yönetme.

    expectation_expr literallardan, tablo içindeki sütun tanımlayıcılarından ve deterministik, yerleşik SQL fonksiyonları veya operatörlerinden oluşabilir, hariç:

    Ayrıca expectation_expr herhangi bir alt sorgu içermemelidir.

  • BÖLÜMLENDİ (sütun [, ...])

    İsteğe bağlı olarak tabloyu sütunların bir alt kümesine göre bölümlere ayırır.

  • CLUSTER BY cümle

    İsteğe bağlı olarak tabloda sıvı kümelemini etkinleştirir. ve PARTITIONED BYöğesini birleştiremezsinizCLUSTER BY.

  • KONUM yolu

    Tablo verileri için isteğe bağlı bir depolama konumu.

  • YORUM tablo_yorumu

    STRING Tabloyu açıklayan değişmez değer.

  • TBLPROPERTIES yan tümcesi

    İsteğe bağlı olarak bir veya daha fazla kullanıcı tanımlı tablo özelliği ayarlar.

  • WITH ROW FILTER yan tümcesi

    Tabloya bir satır filtresi işlevi ekler. Bu tablo için gelecekteki sorgular, işlevinin olarak değerlendirildiği TRUEsatırların bir alt kümesini alır.

  • FLOW INSERT [ONCE] BY NAME sorgusu

    Sonucu query tabloya ekleyen ve sonuç sütunlarını tablo sütunlarıyla ada göre eşleştiren bir ekleme akışı tanımlar. query toplu iş veya akış kaynaklarına başvurabilir. ONCE akışı her güncelleştirme yerine tek bir kez (örneğin, bir geri doldurma) çalıştırır. Her adlandırılmış akış, akış tablosundakiyle aynı FLOW INSERT BY NAME olan işlem hattı güncelleştirmesi başına girişini tam olarak bir kez işler.

  • İTHALAT KONTROL NOKTASI checkpoint_path

    Akışa aktarılmak için mevcut bir Yapılandırılmış Akış kontrol noktasına giden yolu olan bir STRING kelime literal, böylece taşınan akış, kaynağı yeniden işlemek yerine son taahhüt edilen ofsetinden devam eder. Bir kontrol noktası içe aktarmak Beta'da. Bkz. Yapılandırılmış Akış Kontrol Noktasını Taşımak.

Yapılandırılmış Akış kontrol noktasını Göç Et

Important

Bir kontrol noktası içe aktarmak Beta'da.

Mevcut bir Yapılandırılmış Akış iş yükünü kaynağı yeniden işlemeden bir boru hattına taşımak için kullanılır IMPORT CHECKPOINT . Kullandığınız Yapılandırılmış Akış sorgusunuza ayarlayın checkpoint_pathcheckpointLocation ; bu bulut depolama, Unity Kataloğu hacmi veya DBFS yolu olabilir. İlk boru hattı güncellemesinde, akış o kontrol noktasını pipeline'ın yönetilen depolamasına klonlar. Akış, son bağlanmış ofsetten (örneğin toplamalar, deduplication anahtarları ve watermarklar gibi) olduğu gibi devam eder. Sonraki boru hattı güncellemeleri akışın klonlanmış kontrol noktasını kullanır; orijinal kontrol noktası değiştirilmez.

Warning

Pipeline'ı çalıştırmadan önce orijinal Structured Streaming sorgusunu durdurun. Orijinal Yapılandırılmış Akış sorgusu aktarmadan sonra tekrar kullanılabilir, ancak kontrol noktası durumunu yönetmeniz ve boru hattı ile sorgu aynı anda aynı tabloya yazmadığından emin olmalısınız; bu da tekrarlanan veri üretebilir.

Structured Streaming sorgusunu bir boru hattı akışı olarak yeniden oluşturun ve sorgu yazdığı tabloya yazmaya devam eder ve kontrol noktasını içe aktarın:

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

Kontrol noktası yalnızca ilk pipeline güncellemesinde içe aktarılır. Sonraki güncellemeler IMPORT CHECKPOINT. Tam yenileme kontrol noktasını yeniden içe aktarmaz. Yeni, boş bir kontrol noktasından başlıyor ve kaynağı yeniden işliyor.

Sınırlamalar

  • Yönetilen tablolar CDC değişiklik akışlarını desteklemez. AUTO CDC INTOyönetilen tabloya karşı (SQL) veya apply_changes / apply_changes_from_snapshot (Python) ile MANAGED_TABLE_DOES_NOT_SUPPORT_CDCbaşarısız oluyor. CDC hedefleri için, CREATE FLOW (pipelines) ile tanımlanan bir akışa sahip birCREATE STREAMING TABLE (boru hatları) kullanın.
  • Yönetilen tablolar desteklemez FLOW ... REPLACE WHERE. Yalnızca FLOW INSERT BY NAME desteklenir. Bir tablonun hedeflenen alt kümesini yeniden hesaplamak ve üzerine yazmak için, bir akış tablosu oluşturun ve CREATE FLOW (pipelines) ile bir REPLACE WHERE akış tanımlayın. Bkz. REPLACE WHERE akışlarıyla toplu işleme.
  • Yönetilen tablolar yalnızca Unity Kataloğu ile işlem hatlarında desteklenir. Hive meta veri deposu desteklenmez.
  • Yönetilen tablo için mevcut akış tablosunun adını yeniden kullanamazsınız. Önce akış tablosunu bırakın veya deyimi ile CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLEbaşarısız olur.
  • IMPORT CHECKPOINTyalnızca .FLOW INSERT BY NAME

Examples

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

Ek kaynaklar

  • Son güncelleştirme: