Aracılığıyla paylaş

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 [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
      [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...]
    [ , table_constraint ] [...] )

column_properties
  { NOT NULL |
    COMMENT column_comment |
    column_constraint |
    MASK clause } [ ... ]

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

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.

      • MASK yan tümcesi

        Önemli

        Bu özellik Genel Önizlemededir.

        Hassas verileri anonim hale getirmek için bir sütun maskesi işlevi ekler. Bu sütundan gelecek tüm sorgular, sütunun özgün değeri yerine bu işlevi sütun üzerinde değerlendirmenin sonucunu alır. Bu, işlevin, değeri yeniden işlemeye karar vermek için çağıran kullanıcının kimliğini ve/veya grup üyeliklerini incelediği ayrıntılı erişim denetimi amaçları için yararlı olabilir.

      • 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 SCHEDULE CREATE 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.

    • WITH ROW FILTER yan tümcesi

      Önemli

      Bu özellik Genel Önizlemededir.

      Tabloya bir satır filtresi işlevi ekler. Bu tablodan gelecek tüm sorgular, işlevin TRUE değerini değerlendirdiği satırlarının alt kümesini alır. Bu, işlevin belirli satırları filtreleyip filtrelememeye karar vermek için çağıran kullanıcının kimliğini ve/veya grup üyeliklerini incelediği ayrıntılı erişim denetimi amaçları için yararlı olabilir.

  • 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.

Satır filtreleri ve sütun maskeleri

Önemli

Bu özellik Genel Önizlemededir.

Satır filtreleri, tablo taraması satırları her getirdiğinde filtre olarak uygulanan bir işlev belirtmenize olanak tanır. Bu filtreler, izleyen sorguların yalnızca filtre koşulunun true olarak değerlendirildiği satırları döndürmesini sağlar.

Sütun maskeleri, tablo taraması satırları her getirdiğinizde sütunun değerlerini maskelemenizi sağlar. Bu sütunu içeren gelecekteki tüm sorgular, işlevi sütun üzerinde değerlendirerek sütunun özgün değerini değiştirmenin sonucunu alır.

Satır filtrelerini ve sütun maskelerini kullanma hakkında daha fazla bilgi için bkz . Satır filtrelerini ve sütun maskelerini kullanarak hassas tablo verilerini filtreleme.

Satır Filtrelerini ve Sütun Maskelerini Yönetme

Akış tablolarındaki satır filtreleri ve sütun maskeleri deyimine CREATE OR REFRESH eklenmelidir, güncelleştirilmelidir veya bırakılmalıdır.

Davranış

  • Definer olarak yenile: veya REFRESH deyimleri bir akış tablosunu yenilediğindeCREATE OR REFRESH, satır filtresi işlevleri tanımlayıcının haklarıyla (tablo sahibi olarak) çalışır. Bu, tablo yenilemesinin akış tablosunu oluşturan kullanıcının güvenlik bağlamını kullandığı anlamına gelir.
  • Sorgu: Çoğu filtre, tanımlayıcının haklarıyla çalıştırılırken, kullanıcı bağlamını (ve IS_MEMBERgibiCURRENT_USER) denetleen işlevler özel durumlardır. Bu işlevler çağıran olarak çalışır. Bu yaklaşım, geçerli kullanıcının bağlamını temel alarak kullanıcıya özgü veri güvenliği ve erişim denetimlerini zorunlu kılıp uygular.

Gözlemlenebilirlik

Belirli bir akış tablosuna uygulanan mevcut satır filtrelerini ve sütun maskelerini incelemek için , INFORMATION_SCHEMAveya Katalog Gezgini'ni kullanınDESCRIBE EXTENDED. Bu işlevsellik, kullanıcıların akış tablolarındaki veri erişimi ve koruma ölçülerini denetlemesine ve gözden geçirmesine olanak tanır.

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 veya ALTER STREAMING TABLE deyimiyle CREATE OR REFRESH 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');

-- Creates a streaming table with a row filter and a column mask
> CREATE OR REFRESH STREAMING TABLE masked_csv_data (
    id int,
    name string,
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn
  )
  WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
  AS SELECT *
  FROM STREAM read_files('s3://bucket/path/sensitive_data')