Справочник по языку SQL для разностных динамических таблиц

В этой статье содержатся сведения о интерфейсе программирования SQL Delta Live Tables.

• Дополнительные сведения об API Python см. в Справочнике по языку Python для разностных динамических таблиц.
• Дополнительные сведения о командах SQL см . в справочнике по языку SQL.

В запросах SQL можно использовать определяемые пользователем функции Python, но перед их вызовом в исходных файлах SQL необходимо определить эти определяемые пользователем функции Python. См . раздел "Определяемые пользователем скалярные функции " Python".

Ограничения

Предложение PIVOT не поддерживается. Для pivot выполнения операции в Spark требуется загрузка входных данных для вычисления выходной схемы. Эта возможность не поддерживается в разностных динамических таблицах.

Создание материализованного представления или потоковой таблицы Разностных динамических таблиц

При объявлении таблицы потоковой передачи или материализованного представления используется тот же базовый синтаксис SQL.

Объявление материализованного представления разностных динамических таблиц с помощью SQL

Ниже описан синтаксис объявления материализованного представления в разностных динамических таблицах с помощью SQL:

CREATE OR REFRESH MATERIALIZED VIEW view_name [CLUSTER BY (col_name1, col_name2, ... )]
  [(
    [
    col_name1 col_type1 [ GENERATED ALWAYS AS generation_expression1 ] [ COMMENT col_comment1 ] [ column_constraint ] [ MASK func_name [ USING COLUMNS ( other_column_name | constant_literal [, ...] ) ] ],
    col_name2 col_type2 [ GENERATED ALWAYS AS generation_expression2 ] [ COMMENT col_comment2 ] [ column_constraint ] [ MASK func_name [ USING COLUMNS ( other_column_name | constant_literal [, ...] ) ] ],
    ...
    ]
    [
    CONSTRAINT expectation_name_1 EXPECT (expectation_expr1) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    CONSTRAINT expectation_name_2 EXPECT (expectation_expr2) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    ...
    ]
    [ table_constraint ] [, ...]
  )]
  [USING DELTA]
  [PARTITIONED BY (col_name1, col_name2, ... )]
  [LOCATION path]
  [COMMENT table_comment]
  [TBLPROPERTIES (key1 [ = ] val1, key2 [ = ] val2, ... )]
  [ WITH { ROW FILTER func_name ON ( [ column_name | constant_literal [, ...] ] ) [...] } ]
  AS select_statement

Объявление таблицы потоковой передачи разностных динамических таблиц с помощью SQL

Таблицы потоковой передачи можно объявлять только с помощью запросов, которые считываются в источнике потоковой передачи. Databricks рекомендует использовать автозагрузчик для приема файлов из облачного хранилища объектов. См . синтаксис SQL автозагрузчика.

При указании других таблиц или представлений в конвейере в качестве источников потоковой передачи необходимо включить STREAM() функцию вокруг имени набора данных.

Ниже описан синтаксис объявления таблицы потоковой передачи в разностных динамических таблицах с sql:

CREATE OR REFRESH [TEMPORARY] STREAMING TABLE table_name [CLUSTER BY (col_name1, col_name2, ... )]
  [(
    [
    col_name1 col_type1 [ GENERATED ALWAYS AS generation_expression1 ] [ COMMENT col_comment1 ] [ column_constraint ] [ MASK func_name [ USING COLUMNS ( other_column_name | constant_literal [, ...] ) ] ],
    col_name2 col_type2 [ GENERATED ALWAYS AS generation_expression2 ] [ COMMENT col_comment2 ] [ column_constraint ] [ MASK func_name [ USING COLUMNS ( other_column_name | constant_literal [, ...] ) ] ],
    ...
    ]
    [
    CONSTRAINT expectation_name_1 EXPECT (expectation_expr1) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    CONSTRAINT expectation_name_2 EXPECT (expectation_expr2) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    ...
    ]
    [ table_constraint ] [, ...]
  )]
  [USING DELTA]
  [PARTITIONED BY (col_name1, col_name2, ... )]
  [LOCATION path]
  [COMMENT table_comment]
  [TBLPROPERTIES (key1 [ = ] val1, key2 [ = ] val2, ... )]
  [ WITH { ROW FILTER func_name ON ( [ column_name | constant_literal [, ...] ] ) [...] } ]
  AS select_statement

Создание представления разностных динамических таблиц

Ниже описан синтаксис для объявления представлений с помощью SQL:

CREATE TEMPORARY [STREAMING] LIVE VIEW view_name
  [(
    [
    col_name1 [ COMMENT col_comment1 ],
    col_name2 [ COMMENT col_comment2 ],
    ...
    ]
    [
    CONSTRAINT expectation_name_1 EXPECT (expectation_expr1) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    CONSTRAINT expectation_name_2 EXPECT (expectation_expr2) [ON VIOLATION { FAIL UPDATE | DROP ROW }],
    ...
    ]
  )]
  [COMMENT view_comment]
  AS select_statement

Синтаксис автоматического загрузчика SQL

Ниже описан синтаксис для работы с Автозагрузчиком в SQL:

CREATE OR REFRESH STREAMING TABLE table_name
AS SELECT *
  FROM read_files(
    "<file-path>",
    "<file-format>",
    map(
      "<option-key>", "<option_value",
      "<option-key>", "<option_value",
      ...
    )
  )

С Автозагрузчиком можно использовать поддерживаемые параметры формата. map() С помощью функции можно передать параметры методуread_files(). Параметры — это пары "ключ-значение", где ключи и значения являются строками. Дополнительные сведения о форматах и параметрах поддержки см. в параметрах формата файлов.

Пример. Определение таблиц

Набор данных можно создать путем считывания из внешнего источника данных или наборов данных, определенных в конвейере. Для чтения из внутреннего набора данных, добавьте ключевое слово LIVE в начало имени набора данных: В следующем примере определяется два различных набора данных: таблица с именем taxi_raw, которая принимает JSON-файл в качестве источника входных данных, и таблица с именем filtered_data, принимающая таблицу taxi_raw в качестве входных данных:

CREATE OR REFRESH MATERIALIZED VIEW taxi_raw
AS SELECT * FROM json.`/databricks-datasets/nyctaxi/sample/json/`

CREATE OR REFRESH MATERIALIZED VIEW filtered_data
AS SELECT
  ...
FROM LIVE.taxi_raw

Пример. Чтение из источника потоковой передачи

Чтобы считывать данные из источника потоковой передачи, например автозагрузчик или внутренний набор данных, определите таблицу STREAMING :

CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM read_files("/databricks-datasets/retail-org/customers/", "csv")

CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(LIVE.customers_bronze)

Дополнительные сведения о потоковой передаче данных см. в разделе "Преобразование данных с помощью разностных динамических таблиц".

Управление материализацией таблиц

Таблицы также обеспечивают дополнительный контроль над их материализацией:

• Укажите, как секционируются таблицы с помощью PARTITIONED BY. Секционирование можно использовать для ускорения запросов.
• Свойства таблицы можно задать с помощью TBLPROPERTIES. См . свойства таблицы Delta Live Table.
• Задайте место хранения с помощью параметра LOCATION. По умолчанию данные таблицы хранятся в расположении хранилища конвейера, если LOCATION не задано.
• Созданные столбцы можно использовать в определении схемы. См . пример. Указание столбцов схемы и секционирования.

Пример. Указание столбцов схемы и секционирования

При выборе таблицы можно указать схему. В следующем примере указывается схема целевой таблицы, включая использование созданных столбцов Delta Lake и определение столбцов секционирования для таблицы:

CREATE OR REFRESH MATERIALIZED VIEW sales
(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))
) PARTITIONED BY (order_day_of_week)
COMMENT "Raw data on sales"
AS SELECT * FROM ...

По умолчанию разностные динамические таблицы выводят схему из определения table, если схема не указана.

Пример. Определение ограничений таблицы

При указании схемы можно определить первичные и внешние ключи. Ограничения являются информационными и не применяются. См. предложение CONSTRAINT в справочнике по языку SQL.

В следующем примере определяется таблица с ограничением первичного и внешнего ключа:

CREATE OR REFRESH MATERIALIZED VIEW sales
(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)
)
COMMENT "Raw data on sales"
AS SELECT * FROM ...

Параметризация значений, используемых при объявлении таблиц или представлений с помощью SQL

Используется SET для указания значения конфигурации в запросе, объявляющего таблицу или представление, включая конфигурации Spark. Любая таблица или представление, определенные в записной книжке после того, как инструкция SET получила доступ к определенному значению. Любые конфигурации Spark, заданные с помощью инструкции SET, используются при выполнении запроса Spark для любой таблицы или представления после инструкции SET. Чтобы считать значение конфигурации в запросе, используйте синтаксис интерполяции строк ${}. В следующем примере задается значение конфигурации Spark с именем startDate, которое используется в запросе:

SET startDate='2020-01-01';

CREATE OR REFRESH MATERIALIZED VIEW filtered
AS SELECT * FROM src
WHERE date > ${startDate}

Чтобы указать несколько значений конфигурации, используйте отдельную инструкцию SET для каждого значения.

Пример. Определение фильтра строк и маски столбцов

Чтобы создать материализованное представление или таблицу потоковой передачи с фильтром строк и маской столбцов, используйте предложение ROW FILTER и предложение MASK. В следующем примере показано, как определить материализованное представление и таблицу Потоковой передачи с фильтром строк и маской столбца:

CREATE OR REFRESH STREAMING TABLE customers_silver (
  id int COMMENT 'This is the customer ID',
  name string,
  region string,
  ssn string MASK catalog.schema.ssn_mask_fn COMMENT 'SSN masked for privacy'
)
WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
AS SELECT * FROM STREAM(LIVE.customers_bronze)

CREATE OR REFRESH MATERIALIZED VIEW sales (
  customer_id STRING MASK catalog.schema.customer_id_mask_fn,
  customer_name STRING,
  number_of_line_items STRING COMMENT 'Number of items in the order',
  order_datetime STRING,
  order_number LONG,
  order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime))
)
COMMENT "Raw data on sales"
WITH ROW FILTER catalog.schema.order_number_filter_fn ON (order_number)
AS SELECT * FROM LIVE.sales_bronze

Дополнительные сведения о фильтрах строк и масках столбцов см. в статье "Публикация таблиц" с фильтрами строк и масками столбцов.

Свойства SQL

Изменение записи данных с помощью SQL в разностных динамических таблицах

Используйте инструкцию APPLY CHANGES INTO для использования функций CDC Delta Live Table, как описано в следующем разделе:

CREATE OR REFRESH STREAMING TABLE table_name;

APPLY CHANGES INTO LIVE.table_name
FROM source
KEYS (keys)
[IGNORE NULL UPDATES]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2}]
[TRACK HISTORY ON {columnList | * EXCEPT (exceptColumnList)}]

Вы определяете ограничения качества данных для целевого APPLY CHANGES объекта, используя то же CONSTRAINT предложение, что и запросы, отличныеAPPLY CHANGES от запросов. См. статью Управление качеством данных с помощью Delta Live Tables.

См . API APPLY CHANGES: упрощение отслеживания изменений с помощью разностных динамических таблиц.