Share via

AKıŞ TABLOSU OLUŞTURMA

  • Makale

Şunlar için geçerlidir:onay işareti evet olarak işaretlenmiş Databricks SQL onay işareti evet olarak işaretlenmiş Databricks Runtime 13.3 LTS ve üzeri

Önemli

Bu özellik Genel Önizlemededir.

Akış veya artımlı veri işleme için ek destek içeren bir akış tablosu, Delta tablosu oluşturur.

Akış tabloları yalnızca Delta Live Tablolarında ve Unity Kataloğu ile Databricks SQL'de desteklenir. Desteklenen Databricks Runtime işlemlerinde bu komutu çalıştırmak yalnızca söz dizimini ayrıştırmaktadır. Bkz . SQL ile Delta Live Tables işlem hattı uygulama.

Sözdizimi

{ CREATE OR REFRESH STREAMING TABLE | CREATE STREAMING TABLE [ IF NOT EXISTS ] }
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ AS query ]

table_specification
  ( [ column_identifier column_type [ NOT NULL ]
      [ COMMENT column_comment ] [ column_constraint ]
    ] [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
      [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...]
    [ , table_constraint ] [...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    SCHEDULE [ REFRESH ] CRON cron_string [ AT TIME ZONE timezone_id ] } [...]

Parametreler

  • REFRESH

    Belirtilirse, tabloyu sorguda tanımlanan kaynaklardan sağlanan en son verilerle yeniler. Yalnızca sorgu başlamadan önce gelen yeni veriler işlenir. Komutun yürütülmesi sırasında kaynaklara eklenen yeni veriler bir sonraki yenilemeye kadar yoksayılır.

  • YOKSA

    Belirtilirse ve aynı ada sahip bir tablo zaten varsa, deyimi yoksayılır.

    IF NOT EXISTS ile REFRESHbirlikte kullanılamaz, bu da izin verilmediği anlamına gelir CREATE OR REFRESH TABLE IF NOT EXISTS .

  • Table_name

    Oluşturulacak tablonun adı. Ad bir zamansal belirtim içermemelidir. Ad uygun değilse, tablo geçerli şemada oluşturulur.

  • table_specification

    Bu isteğe bağlı yan tümcesi sütunların listesini, türlerini, özelliklerini, açıklamalarını ve sütun kısıtlamalarını tanımlar.

    Tablo şemasında sütun tanımlamazsanız belirtmelisiniz AS query.

    • column_identifier

      Sütun için benzersiz bir ad.

      • column_type

        Sütunun veri türünü belirtir.

      • NOT NULL

        Belirtilirse, sütun değerleri kabul NULL etmez.

      • COMMENT column_comment

        Sütunu açıklamak için bir dize değişmez değeri.

      • column_constraint

        Önemli

        Bu özellik Genel Önizlemededir.

        Akış tablosundaki sütuna birincil anahtar veya yabancı anahtar kısıtlaması ekler. Kısıtlamalar katalogdaki hive_metastore tablolar için desteklenmez.

      • CONSTRAINT expectation_name EXPECT (expectation_expr) [ İHLALDE { FAIL UPDATE | DROP ROW } ]

        Tabloya veri kalitesi beklentileri ekler. Bu veri kalitesi beklentileri zaman içinde izlenebilir ve akış tablosunun olay günlüğü aracılığıyla erişilebilir. Beklenti FAIL UPDATE , hem tabloyu oluştururken hem de tabloyu yenilerken işlemenin başarısız olmasına neden olur. Beklenti DROP ROW , beklenti karşılanmazsa satırın tamamının bırakılmasına neden olur.

        expectation_expr değişmez değerlerden, tablo içindeki sütun tanımlayıcılarından ve aşağıdakiler dışında belirleyici, yerleşik SQL işlevlerinden veya işleçlerinden oluşabilir:

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

      • table_constraint

        Önemli

        Bu özellik Genel Önizlemededir.

        Bir akış tablosuna bilgi birincil anahtarı veya bilgilendiren yabancı anahtar kısıtlamaları ekler. Katalogdaki hive_metastore tablolar için anahtar kısıtlamaları desteklenmez.

  • table_clauses

    İsteğe bağlı olarak bölümleme, açıklamalar, kullanıcı tanımlı özellikler ve yeni tablo için yenileme zamanlaması belirtin. Her alt yan tümce yalnızca bir kez belirtilebilir.

    • BÖLÜMLENDİ

      Tabloyu bölümleme ölçütü olarak tablo sütunlarının isteğe bağlı listesi.

    • COMMENT table_comment

      STRING Tabloyu açıklamak için bir değişmez değer.

    • TBLPROPERTIES

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

    • SCHEDULE [ REFRESH ] CRON cron_string [ TIME ZONE timezone_id ]

      Sağlandıysa, akış tablosunu veya gerçekleştirilmiş görünümü zamanlayarak verilerini verilen quartz cron zamanlaması ile yeniler. Yalnızca time_zone_values kabul edilir. AT TIME ZONE LOCAL Desteklenmez. Yoksa AT TIME ZONE , oturum saat dilimi kullanılır. Yoksa AT TIME ZONE ve oturum saat dilimi ayarlanmadıysa bir hata oluşur. SCHEDULE , ile eşanlamlı olarak eşdeğerdir SCHEDULE REFRESH.

      Delta Live Tables işlem hattı tanımında söz dizimini kullanamazsınız SCHEDULE .

      Komutta SCHEDULECREATE OR REFRESH yan tümcesine izin verilmez. Zamanlama komutun CREATE bir parçası olarak sağlanabilir. Oluşturma işleminden sonra akış tablosunun zamanlamasını değiştirmek için ALTER STREAMING TABLE kullanın.

  • AS sorgusu

    Bu yan tümce, içindeki verileri querykullanarak tabloyu doldurur. Bu sorgu bir akış sorgusu olmalıdır. Bu, artımlı olarak işlemek istediğiniz herhangi bir ilişkiye anahtar sözcüğünü ekleyerek STREAM elde edilebilir. birlikte bir query ve table_specification belirttiğinizde, içinde table_specification belirtilen tablo şeması tarafından querydöndürülen tüm sütunları içermelidir, aksi takdirde bir hata alırsınız. içinde table_specification belirtilen ancak sorgulandığında döndürülen null değerler tarafından query döndürülmeyen tüm sütunlar.

    Bu yan tümce Databricks SQL'de oluşturulan akış tabloları için gereklidir, ancak Delta Live Tablolarında gerekli değildir. Delta Live Tables'da bu yan tümce sağlanmazsa, DLT işlem hattınızdaki bir APPLY CHANGES komutta bu tabloya başvurmanız gerekir. Bkz. Delta Live Tablolarında SQL ile veri yakalamayı değiştirme.

Akış tabloları ile diğer tablolar arasındaki farklar

Akış tabloları, büyüyen bir veri kümesini işlerken her satırı yalnızca bir kez işlemek üzere tasarlanmış durum bilgisi olan tablolardır. Çoğu veri kümesi zaman içinde sürekli büyüdüğü için akış tabloları çoğu alım iş yükü için iyidir. Akış tabloları, veri güncelliği ve düşük gecikme süresi gerektiren işlem hatları için idealdir. Akış tabloları, yeni veriler geldikçe artımlı olarak hesaplanabilir ve her güncelleştirmede tüm kaynak verileri tam olarak yeniden derlemeye gerek kalmadan sonuçları güncel tutarak büyük ölçekli dönüşümler için de yararlı olabilir. Akış tabloları yalnızca ekli veri kaynakları için tasarlanmıştır.

Akış tabloları, sorguda sağlanan kaynaklarda bulunan en son verileri işleyen gibi REFRESHek komutları kabul eder. Sağlanan sorguda yapılan değişiklikler yalnızca daha önce işlenen verileri değil çağrısı REFRESHyaparak yeni verilere yansıtılır. Değişiklikleri var olan verilere de uygulamak için bir FULL REFRESHgerçekleştirmek için komutunu yürütmeniz REFRESH TABLE <table_name> FULL gerekir. Tam yenilemeler, kaynakta bulunan tüm verileri en son tanım ile yeniden işler. Verilerin geçmişinin tamamını tutmayan veya Kafka gibi kısa saklama süreleri olan kaynaklarda tam yenilemelerin çağrılması önerilmez çünkü tam yenileme mevcut verileri kısaltmaktadır. Veriler artık kaynakta kullanılamıyorsa eski verileri kurtaramayabilirsiniz.

Sınırlamalar

  • En son verileri almak için yalnızca tablo sahipleri akış tablolarını yenileyebilir.

  • ALTER TABLE akış tablolarında komutlara izin verilmiyor. Tablonun tanımı ve özellikleri deyimiyle ALTER STREAMING TABLE değiştirilmelidir.

  • Zaman yolculuğu sorguları desteklenmez.

  • tablo şemasını gibi INSERT INTODML komutları aracılığıyla geliştirme ve MERGE desteklenmez.

  • Akış tablolarında aşağıdaki komutlar desteklenmez:

    • CREATE TABLE ... CLONE <streaming_table>
    • COPY INTO
    • ANALYZE TABLE
    • RESTORE
    • TRUNCATE
    • GENERATE MANIFEST
    • [CREATE OR] REPLACE TABLE

  • Delta Paylaşımı desteklenmez.

  • Tabloyu yeniden adlandırmak veya sahibi değiştirmek desteklenmez.

  • ve FOREIGN KEY gibi PRIMARY KEY tablo kısıtlamaları desteklenmez.

  • Oluşturulan sütunlar, kimlik sütunları ve varsayılan sütunlar desteklenmez.

Örnekler

-- Creates a streaming table that processes files stored in the given external location with
-- schema inference and evolution.
> CREATE OR REFRESH STREAMING TABLE raw_data
  AS SELECT * FROM STREAM read_files('abfss://container@storageAccount.dfs.core.windows.net/base/path');

-- Creates a streaming table that processes files with a known schema.
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with schema evolution and data quality expectations.
-- The table creation or refresh fails if the data doesn't satisfy the expectation.
> CREATE OR REFRESH STREAMING TABLE avro_data (
    CONSTRAINT date_parsing (to_date(dt) >= '2000-01-01') ON VIOLATION FAIL UPDATE
  )
  AS SELECT *
  FROM STREAM read_files('gs://my-bucket/avroData');

-- Stores the data from Kafka in an append-only streaming table.
> CREATE OR REFRESH STREAMING TABLE firehose_raw
  COMMENT 'Stores the raw data from Kafka'
  TBLPROPERTIES ('delta.appendOnly' = 'true')
  AS SELECT
    value raw_data,
    offset,
    timestamp,
    timestampType
  FROM STREAM read_kafka(bootstrapServers => 'ips', subscribe => 'topic_name');

-- Read data from another streaming table scheduled to run every hour.
> CREATE STREAMING TABLE firehose_bronze
  SCHEDULE CRON '0 0 * * * ? *'
  AS SELECT
    from_json(raw_data, 'schema_string') data,
    * EXCEPT (raw_data)
  FROM STREAM firehose_raw;

-- Creates a streaming table with a column constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int PRIMARY KEY,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with a table constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string,
    CONSTRAINT pk_id PRIMARY KEY (id)
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');