Как переместить или изменить трубу без потери или дублирования каких-либо записей

Question

На этой странице по управлению конвейерами предлагается процесс преобразования копии в оператор конвейера.

1. Приостановите конвейер (используя ALTER PIPE … SET PIPE_EXECUTION_PAUSED=true).
2. Выполните запрос к функции SYSTEM$PIPE_STATUS и убедитесь, что состояние выполнения канала - PAUSED, а количество ожидающих файлов равно 0.
3. Восстановите канал, чтобы изменить оператор COPY в определении. Выберите один из следующих вариантов: Отбросьте трубу (используя DROP PIPE) и создайте ее (используя CREATE PIPE). Восстановите трубу (используя синтаксис CREATE OR REPLACE PIPE). Внутри труба отбрасывается и создается.
4. Снова запросите функцию SYSTEM$PIPE_STATUS и убедитесь, что состояние выполнения канала - RUNNING.

Однако, если файл должен быть загружен во время между приостановкой и воссозданием канала, здесь нет шага для обновления для этого промежутка. Даже если эти шаги выполняются быстро, у нас есть примеры пропуска файлов.

Запуск ALTER PIPE REFRESHтем не менее, вызывает дублирование, потому что история копирования привязана к каналу. У воссозданной трубы нет этой истории, и она вернется и перезагрузит все.

Есть ли хороший способ записать подобное изменение, чтобы гарантировать отсутствие пропусков или перекрытий? Что-то вроде получения отметки времени, когда исходный канал был приостановлен, а затем использования этой отметки времени в запросе на обновление?

Обновление: мы создали полный процесс и комбинацию скриптов для обработки наших сценариев. Полный сценарий включен в ответ ниже.

Answer 1

Мы написали этот процесс / скрипт для управления снежными трубами. Он включает проверку и обновление файлов, которые были отброшены при изменении или перемещении канала.

Это несколько специфично для наших процессов и стандартов, но должно быть легко изменено / обобщено в зависимости от варианта использования.

/* =====================================================================
This script provides tools for managing, altering, migrating snowpipes
    and their underlying tables. These instructions may need to be tweaked
    depending on your exact use case.

Assuming the pipes and stages follow our standard naming conventions,
    you can find and replace <Database_Name>, <Schema_Name>, <Table_Name>
    with their respective values

======================================================================== */

------------------------------------------
-- Set up Context and Variables
------------------------------------------
--Set your context so you don’t accidently run scripts in the wrong place
use <Database_Name>.<Schema_Name>

--Pause the pipe, also giving us the pipe altered time
--Prefer to avoid running this at the top of the hour to make it easier to verify file loads below.
alter pipe <Table_Name>_PIPE set pipe_execution_paused = true;

--Get timestamp for when pipe was paused and set as variable $pipe_altered_time
set pipe_altered_time = (select last_altered::timestamp_tz from information_schema.pipes where pipe_name = '<Table_Name>_PIPE');

--View variable from above to verify that it was altered just now
select $pipe_altered_time;

------------------------------------------
-- Alter table as per request
------------------------------------------
--This step will change depending on the resquest. E.g. it can be skipped if you are just moving a pipe

    --Example for adding columns to a pipe
    alter table <Table_Name> add column <CREATE_DATETIME_UTC> <timestamp_NTZ>;
    alter table <Table_Name> add column <INCOMING_RAW_REQUEST_ID> <STRING>;
    alter table <Table_Name> add column <INITIAL_PING_ID> <INT>;
    --Viewing if new column was added
    --Note: Make sure the order of all of the columns matches the pipe column order below
    select top 10 * from <Table_Name>;

    --Example for migrating a pipe
    create table <New_Table_Name> clone <Old_Database_Name>.<Old_Schema_Name>.<Old_Table_Name>

------------------------------------------
-- Alter Pipe
------------------------------------------
--Altered pipe statement and turns the pipe on
--This DDL should be retrieved using `SELECT GET_DDL('pipe', '<Table_Name>_PIPE')` (in the same schema context) and then modified to add/adjust columns if needed.
--Make sure that extracted column names here come from the JSON attribute names, which should be provided in the request.
--Most Pipes should follow a very similar pattern to the example below
create or replace pipe <Table_Name>_PIPE
    auto_ingest=true 
    integration='EVENT_HUB_QUEUE' 
    as 
<
   COPY INTO <Table_Name>
    FROM (SELECT PARSE_JSON(HEX_DECODE_STRING($1:Body)) as JSON, 1 as IS_ACTIVE, metadata$filename, JSON:CreatedDateUtc, JSON:Id, JSON:InitialPingId FROM @<Table_Name>_STAGE)
>;

--"Refresh" pipe. Reprocess files that were loaded during the paused time frame. 
-- This can take a few minutes depending on the table and how long it was paused.
alter pipe <Table_Name>_PIPE REFRESH MODIFIED_AFTER = $pipe_altered_time;

-------------------------------------------------------------------------------------------------------
--VALIDATION
-------------------------------------------------------------------------------------------------------

------------------------------------------
-- Verify that new files are being processed as expected
------------------------------------------

--Data validation - see what files have been loaded
--It can take a while for new files to show up depending on the event_hub.
--The real thing you want to check is that the pipe is working, and you see files from after the time when the pipe was paused
--If there are new files coming in from Event_hub, but not making it to the table, the pipe may have errors that will help for troubleshooting
select LAST_LOAD_TIME, ROW_COUNT, FILE_NAME
  from table(information_schema.copy_history(Table_Name=>'<Table_Name>', start_time=> dateadd(hours, -2, current_timestamp())))
  ORDER BY LAST_LOAD_TIME DESC;

--Validate data made it to the new columns
--Make sure that there are new records since the $pipe_altered_time in the table (it might take a while for them to show up)
--  Validate that the new records look as expected
--  If there is confusion here, it would be appropriate to contact the requestor on if the data looks correct
select * from <Table_Name>
where <any new column name> is not null;

------------------------------------------
--compare files in the storage account vs files loaded into the table
------------------------------------------

--Choose an hour of the day for comparing (Generally the hour during which the pipe was paused ($pipe_altered_time))
--see what files have been loaded recently if needed
select distinct top 1000 file_name from <Table_Name>
order by file_name desc;

--run these to create variables
--Remove the 'hh' from the directory format below to query an entire day.
set hour_to_check = DATE_TRUNC('Hour', $pipe_altered_time)::timestamp_ntz;
set date_directory = to_varchar($hour_to_check, '/YYYY/MM/DD/hh');
set file_pattern = '''.*' || $date_directory || '.*''';
select $file_pattern;

-- Get the list of files in the storage account. 
-- This query can take 10+ minutes for storage accounts with many files
-- Note: This is currently not dynamic, and doesn't like variables. You need to manually enter the value from $file_pattern variable above
LIST @<Table_Name>_STAGE pattern= <'.*/2020/04/23/17.*'>;

-- Comparison stuff
-- Uses the output of the previous query and compares it to what's been loaded in the table
SELECT s.$1 as azurefilename, t.file_name as tablefilename
FROM
(
    table(RESULT_SCAN(LAST_QUERY_ID()))
    --If you need to specify the query id for the LIST results manually, use this instead
    --table(RESULT_SCAN('<abcabcab-00bd-74d1-0000-1d990c3c72aa>'))
) s
FULL OUTER JOIN 
(
    select distinct file_name from OUTGOING_RAW_REQUEST
    where file_name like '%' || $date_directory || '%'
) t
ON charindex(t.file_name, s.$1) > 0;

-- Check for mismatched files
-- Ideally, this list should be empty
select *
    from 
    (
        table(RESULT_SCAN(LAST_QUERY_ID()))
        --If you need to specify the query id for the comparison results manually, use this instead
        --table(RESULT_SCAN('<abcabcab-00bd-74d1-0000-1d990c3c72aa>'))
    )
    where $1 is null or $2 is null;

-- If the above list is empty. You're done.
-- If the above list returns files that were not loaded, then copy the missing files into table.
-- Get the copy into statement. From the pipe creation statement above or using `select get_ddl('pipe', '<pipe_name>')`
--      You'll need to cut out the pipe syntax and just use the `COPY INTO` statement
-- the file name format should generally look like `/2020/04/23/17/20/53/28.avro` and you may need to trim the results from above to match.
<COPY INTO OUTGOING_RAW_REQUEST FROM (SELECT PARSE_JSON(HEX_DECODE_STRING($1:Body)), '1'::numeric, metadata$filename FROM @OUTGOING_RAW_REQUEST_STAGE)>
files = (
'</2020/04/23/17/20/53/28.avro>',
'</2020/04/23/17/20/52/20.avro>');

Answer 2

К сожалению, это текущее ограничение для автоматического захвата Snowpipe, и нет отличного обходного пути.

Вы абсолютно правы в том, что история привязана к определенной трубе. Когда труба воссоздана, технически у вас есть новая труба с новой историей. Таким образом, ALTER PIPE REFRESH снова перезагрузит все с указанным префиксом, что приведет к дублированию данных.

Вы можете использовать опцию "MODIFIED_DATE" для ALTER PIPE REFRESH и указать время, равное времени последнего pipe_recieve_time, но вы можете пропустить некоторые уведомления, поскольку входящий заказ не гарантирован.

Чтобы свести к минимуму количество повторяющихся данных, вы можете организовать свои файлы данных в более детальной структуре даты и времени (например, год / месяц / день / час / минута /...) и обновлять только папки, соответствующие времени, в котором находилась ваша труба. "вниз". Однако могут быть некоторые файлы, которые уже были загружены предыдущим конвейером, что приводит к дублированию данных.