• Описание фреймворка MDW и работ по его донастройке под GreenPlum
• Сравнительные характеристики производительности Ab Initio по работе с Hive и GreenPlum
• Работа Ab Initio с GreenPlum в режиме Near Real Time

Описание Ab Initio

Описание фреймворка MDW и работ по его донастройке под GreenPlum

distributed by

delete <table> all

delete from <table>

delete from t where rowid in (<соединение t с дельтой>)

delete from t where exists (select * from delta where delta.pk=t.pk)

where ABLOCAL()

ablocal_expr=string_concat("mod(t.", string_filter_out("{$TABLE_KEY}","{}"), ",", (decimal(3))(number_of_partitions()),")=", (decimal(3))(this_partition()))

mod(sk,10)=3

Сравнительные характеристики производительности Ab Initio по работе с Hive и GreenPlum

• инициализирующая загрузка + инкрементальная загрузка случайно сгенерированных данных в таблицу Hive
• инициализирующая загрузка + инкрементальная загрузка случайно сгенерированных данных в такую же таблицу GreenPlum

Работа Ab Initio с GreenPlum в режиме Near Real Time

• сгенерировать 6 000 000 тестовых строк
• произвести инициализирующую загрузку вставить 6 000 000 тестовых строк в пустую таблицу
• повторить 5 раз инкрементальную загрузку
  
  • сгенерировать 6 000 000 тестовых строк
  • произвести инкрементальную вставку 6 000 000 тестовых строк в таблицу (при этом старым данным проставляется время истечения актуальности valid_to_ts и вставляются более свежие данные с тем же первичным ключом)

select valid_from_ts, valid_to_ts, count(1), min(sk), max(sk) from dev42_1_db_usl.TESTING_SUBJ_org_finval group by valid_from_ts, valid_to_ts order by 1,2;

Заключение

Автор эксперт профессионального сообщества Сбербанка SberProfi DWH/BigData. Профессиональное сообщество SberProfi DWH/BigData отвечает за развитие компетенций в таких направлениях, как экосистема Hadoop, Teradata, Oracle DB, GreenPlum, а также BI инструментах Qlik, SAP BO, Tableau и др.

• Автоматическое журналирование компиляций
• Как быть, если хочется сделать вьюшку с параметрами
• Использование динамической статистики в запросах
• Как сохранить план запроса при вставке данных через database link
• Запуск процедур в параллельных сессиях
• Протягивание остатков
• Объединение нескольких историй в одну
• Нормалайзер
• Визуализация в формате SVG
• Приложение поиска по метаданным Oracle

Автоматическое журналирование компиляций

create table DDL_CHANGES_LOG(  id               INTEGER,  change_date      DATE,  sid              VARCHAR2(100),  schemaname       VARCHAR2(30),  machine          VARCHAR2(100),  program          VARCHAR2(100),  osuser           VARCHAR2(100),  obj_owner        VARCHAR2(30),  obj_type         VARCHAR2(30),  obj_name         VARCHAR2(30),  previous_version CLOB,  changes_script   CLOB);create or replace trigger trig_audit_ddl_trg  before ddl on databasedeclare  v_sysdate              date;  v_valid                number;  v_previous_obj_owner   varchar2(30) := '';  v_previous_obj_type    varchar2(30) := '';  v_previous_obj_name    varchar2(30) := '';  v_previous_change_date date;  v_lob_loc_old          clob := '';  v_lob_loc_new          clob := '';  v_n                    number;  v_sql_text             ora_name_list_t;  v_sid                  varchar2(100) := '';  v_schemaname           varchar2(30) := '';  v_machine              varchar2(100) := '';  v_program              varchar2(100) := '';  v_osuser               varchar2(100) := '';begin  v_sysdate := sysdate;  -- find whether compiled object already presents and is valid  select count(*)    into v_valid    from sys.dba_objects   where owner = ora_dict_obj_owner     and object_type = ora_dict_obj_type     and object_name = ora_dict_obj_name     and status = 'VALID'     and owner not in ('SYS', 'SPOT', 'WMSYS', 'XDB', 'SYSTEM')     and object_type in ('TRIGGER', 'PROCEDURE', 'FUNCTION', 'PACKAGE', 'PACKAGE BODY', 'VIEW');  -- find information about previous compiled object  select max(obj_owner) keep(dense_rank last order by id),         max(obj_type) keep(dense_rank last order by id),         max(obj_name) keep(dense_rank last order by id),         max(change_date) keep(dense_rank last order by id)    into v_previous_obj_owner, v_previous_obj_type, v_previous_obj_name, v_previous_change_date    from ddl_changes_log;  -- if compile valid object or compile invalid package body broken by previous compilation of package then log it  if (v_valid = 1 or v_previous_obj_owner = ora_dict_obj_owner and     (v_previous_obj_type = 'PACKAGE' and ora_dict_obj_type = 'PACKAGE BODY' or     v_previous_obj_type = 'PACKAGE BODY' and ora_dict_obj_type = 'PACKAGE') and     v_previous_obj_name = ora_dict_obj_name and     v_sysdate - v_previous_change_date <= 1 / 24 / 60 / 2) and     ora_sysevent in ('CREATE', 'ALTER') then    -- store previous version of object (before compilation) from dba_source or dba_views in v_lob_loc_old    if ora_dict_obj_type <> 'VIEW' then      for z in (select substr(text, 1, length(text) - 1) || chr(13) || chr(10) as text                  from sys.dba_source                 where owner = ora_dict_obj_owner                   and type = ora_dict_obj_type                   and name = ora_dict_obj_name                 order by line) loop        v_lob_loc_old := v_lob_loc_old || z.text;      end loop;    else      select sys.dbms_metadata_util.long2clob(v.textlength, 'SYS.VIEW$', 'TEXT', v.rowid) into v_lob_loc_old        from sys."_CURRENT_EDITION_OBJ" o, sys.view$ v, sys.user$ u       where o.obj# = v.obj#         and o.owner# = u.user#         and u.name = ora_dict_obj_owner         and o.name = ora_dict_obj_name;    end if;    -- store new version of object (after compilation) from v_sql_text in v_lob_loc_new    v_n := ora_sql_txt(v_sql_text);    for i in 1 .. v_n loop      v_lob_loc_new := v_lob_loc_new || replace(v_sql_text(i), chr(10), chr(13) || chr(10));    end loop;    -- find information about session that changed this object    select max(to_char(sid)), max(schemaname), max(machine), max(program), max(osuser)      into v_sid, v_schemaname, v_machine, v_program, v_osuser      from v$session     where audsid = userenv('sessionid');    -- store changes in ddl_changes_log    insert into ddl_changes_log      (id, change_date, sid, schemaname, machine, program, osuser,       obj_owner, obj_type, obj_name, previous_version, changes_script)    values      (seq_ddl_changes_log.nextval, v_sysdate, v_sid, v_schemaname, v_machine, v_program, v_osuser,       ora_dict_obj_owner, ora_dict_obj_type, ora_dict_obj_name, v_lob_loc_old, v_lob_loc_new);  end if;exception  when others then    null;end;

Как быть, если хочется сделать вьюшку с параметрами

create table DIVISION_SALES(  division_id INTEGER,  dt          DATE,  sales_amt   NUMBER);

select t1.division_id,       t1.dt          dt1,       t2.dt          dt2,       t1.sales_amt   sales_amt1,       t2.sales_amt   sales_amt2  from (select dt, division_id, sales_amt          from division_sales         where dt = to_date('30.04.2020', 'dd.mm.yyyy')) t1,       (select dt, division_id, sales_amt          from division_sales         where dt = to_date('11.09.2020', 'dd.mm.yyyy')) t2 where t1.division_id = t2.division_id;

create or replace view vw_division_sales_report(in_dt1 date, in_dt2 date) asselect t1.division_id,       t1.dt          dt1,       t2.dt          dt2,       t1.sales_amt   sales_amt1,       t2.sales_amt   sales_amt2  from (select dt, division_id, sales_amt          from division_sales         where dt = in_dt1) t1,       (select dt, division_id, sales_amt          from division_sales         where dt = in_dt2) t2 where t1.division_id = t2.division_id;

create type t_division_sales_report as object(  division_id INTEGER,  dt1         DATE,  dt2         DATE,  sales_amt1  NUMBER,  sales_amt2  NUMBER);

create type t_division_sales_report_table as table of t_division_sales_report;

create or replace function func_division_sales(in_dt1 date, in_dt2 date)  return t_division_sales_report_table  pipelined asbegin  for z in (select t1.division_id,                   t1.dt          dt1,                   t2.dt          dt2,                   t1.sales_amt   sales_amt1,                   t2.sales_amt   sales_amt2              from (select dt, division_id, sales_amt                      from division_sales                     where dt = in_dt1) t1,                   (select dt, division_id, sales_amt                      from division_sales                     where dt = in_dt2) t2             where t1.division_id = t2.division_id) loop    pipe row(t_division_sales_report(z.division_id,                                     z.dt1,                                     z.dt2,                                     z.sales_amt1,                                     z.sales_amt2));  end loop;end;

select *  from table(func_division_sales(to_date('30.04.2020', 'dd.mm.yyyy'),                                 to_date('11.09.2020', 'dd.mm.yyyy')));

create or replace view complex_view as select field1, ...   from (select field1, ...           from (select field1, ... from deep_table), table1          where ...),        table2  where ...;

select field1, ... from complex_view where field1 = 'myvalue';

Использование динамической статистики в запросах

SQL> EXPLAIN PLAN  2   SET statement_id = 'test1'  3   INTO plan_table  4  FOR  with c as  5   (select /*+ dynamic_sampling(change_balances 2)*/  6     account_id, balance_amount  7      from change_balances  8     where mod(account_id, 10) = 0)  9  select a.account_id, a.account_number, c.balance_amount 10    from c, accounts a 11   where c.account_id = a.account_id;Explained.SQL>SQL> SELECT * FROM table (DBMS_XPLAN.DISPLAY);Plan hash value: 874320301----------------------------------------------------------------------------------------------| Id  | Operation          | Name            | Rows  | Bytes |TempSpc| Cost (%CPU)| Time     |----------------------------------------------------------------------------------------------|   0 | SELECT STATEMENT   |                 |  9951K|   493M|       |   140K  (1)| 00:28:10 ||*  1 |  HASH JOIN         |                 |  9951K|   493M|  3240K|   140K  (1)| 00:28:10 ||*  2 |   TABLE ACCESS FULL| CHANGE_BALANCES |   100K|  2057K|       |  7172   (1)| 00:01:27 ||   3 |   TABLE ACCESS FULL| ACCOUNTS        |    10M|   295M|       |   113K  (1)| 00:22:37 |----------------------------------------------------------------------------------------------Predicate Information (identified by operation id):---------------------------------------------------   1 - access("ACCOUNT_ID"="A"."ACCOUNT_ID")   2 - filter(MOD("ACCOUNT_ID",10)=0)Note-----   - dynamic sampling used for this statement (level=2)20 rows selected.

SQL> EXPLAIN PLAN  2   SET statement_id = 'test2'  3   INTO plan_table  4  FOR  with c as  5   (select /*+ dynamic_sampling(change_balances 2)*/  6     account_id, balance_amount  7      from change_balances  8     where mod(account_id, 1000) = 0)  9  select a.account_id, a.account_number, c.balance_amount 10    from c, accounts a 11   where c.account_id = a.account_id;Explained.SQL>SQL> SELECT * FROM table (DBMS_XPLAN.DISPLAY);Plan hash value: 2360715730-------------------------------------------------------------------------------------------------------| Id  | Operation                    | Name                   | Rows  | Bytes | Cost (%CPU)| Time     |-------------------------------------------------------------------------------------------------------|   0 | SELECT STATEMENT             |                        | 73714 |  3743K| 16452   (1)| 00:03:18 ||   1 |  NESTED LOOPS                |                        |       |       |            |          ||   2 |   NESTED LOOPS               |                        | 73714 |  3743K| 16452   (1)| 00:03:18 ||*  3 |    TABLE ACCESS FULL         | CHANGE_BALANCES        |   743 | 15603 |  7172   (1)| 00:01:27 ||*  4 |    INDEX RANGE SCAN          | IX_ACCOUNTS_ACCOUNT_ID |   104 |       |     2   (0)| 00:00:01 ||   5 |   TABLE ACCESS BY INDEX ROWID| ACCOUNTS               |    99 |  3069 |   106   (0)| 00:00:02 |-------------------------------------------------------------------------------------------------------Predicate Information (identified by operation id):---------------------------------------------------   3 - filter(MOD("ACCOUNT_ID",1000)=0)   4 - access("ACCOUNT_ID"="A"."ACCOUNT_ID")Note-----   - dynamic sampling used for this statement (level=2)22 rows selected.

Как сохранить план запроса при вставке данных через database link

insert into dwh_table  (field1, field2)  select field1, field2 from vw_for_dwh_table@xe_link;

SQL> EXPLAIN PLAN  2   SET statement_id = 'test'  3   INTO plan_table  4  FOR  insert into dwh_table  5    (field1, field2)  6    select field1, field2 from vw_for_dwh_table@xe_link;Explained.SQL>SQL> SELECT * FROM table (DBMS_XPLAN.DISPLAY);Plan hash value: 1788691278-------------------------------------------------------------------------------------------------------------| Id  | Operation                | Name             | Rows  | Bytes | Cost (%CPU)| Time     | Inst   |IN-OUT|-------------------------------------------------------------------------------------------------------------|   0 | INSERT STATEMENT         |                  |     1 |  2015 |     2   (0)| 00:00:01 |        |      ||   1 |  LOAD TABLE CONVENTIONAL | DWH_TABLE        |       |       |            |          |        |      ||   2 |   REMOTE                 | VW_FOR_DWH_TABLE |     1 |  2015 |     2   (0)| 00:00:01 | XE_LI~ | R->S |-------------------------------------------------------------------------------------------------------------Remote SQL Information (identified by operation id):----------------------------------------------------   2 - SELECT /*+ OPAQUE_TRANSFORM */ "FIELD1","FIELD2" FROM "VW_FOR_DWH_TABLE" "VW_FOR_DWH_TABLE"       (accessing 'XE_LINK' )16 rows selected.

declare  cursor cr is    select field1, field2 from vw_for_dwh_table@xe_link;  cr_row cr%rowtype;begin  open cr;  loop    fetch cr      into cr_row;    insert into dwh_table      (field1, field2)    values      (cr_row.field1, cr_row.field2);    exit when cr%notfound;  end loop;  close cr;end;

select field1, field2 from vw_for_dwh_table@xe_link;

insert into dwh_table  (field1, field2)  select field1, field2 from vw_for_dwh_table@xe_link;

Запуск процедур в параллельных сессиях

create table PARALLEL_PROC_GROUP_LIST(  group_id   INTEGER,  group_name VARCHAR2(4000));comment on column PARALLEL_PROC_GROUP_LIST.group_id  is 'Номер группы параллельно запускаемых процедур';comment on column PARALLEL_PROC_GROUP_LIST.group_name  is 'Название группы параллельно запускаемых процедур';

create table PARALLEL_PROC_LIST(  group_id    INTEGER,  proc_script VARCHAR2(4000),  is_active   CHAR(1) default 'Y');comment on column PARALLEL_PROC_LIST.group_id  is 'Номер группы параллельно запускаемых процедур';comment on column PARALLEL_PROC_LIST.proc_script  is 'Pl/sql блок с кодом процедуры';comment on column PARALLEL_PROC_LIST.is_active  is 'Y - active, N - inactive. С помощью этого поля можно временно отключать процедуру из группы';

create table PARALLEL_PROC_LOG(  run_id      INTEGER,  group_id    INTEGER,  proc_script VARCHAR2(4000),  job_id      INTEGER,  start_time  DATE,  end_time    DATE);comment on column PARALLEL_PROC_LOG.run_id  is 'Номер запуска процедуры run_in_parallel';comment on column PARALLEL_PROC_LOG.group_id  is 'Номер группы параллельно запускаемых процедур';comment on column PARALLEL_PROC_LOG.proc_script  is 'Pl/sql блок с кодом процедуры';comment on column PARALLEL_PROC_LOG.job_id  is 'Job_id джоба, в котором была запущена эта процедура';comment on column PARALLEL_PROC_LOG.start_time  is 'Время начала работы';comment on column PARALLEL_PROC_LOG.end_time  is 'Время окончания работы';create sequence Seq_Parallel_Proc_Log;

create or replace procedure run_in_parallel(in_group_id integer) as  -- Процедура по параллельному запуску процедур из таблицы parallel_proc_list.  -- Параметр - номер группы из parallel_proc_list  v_run_id             integer;  v_job_id             integer;  v_job_id_list        varchar2(32767);  v_job_id_list_ext    varchar2(32767);  v_running_jobs_count integer;begin  select seq_parallel_proc_log.nextval into v_run_id from dual;  -- submit jobs with the same parallel_proc_list.in_group_id  -- store seperated with ',' JOB_IDs in v_job_id_list  v_job_id_list     := null;  v_job_id_list_ext := null;  for z in (select pt.proc_script              from parallel_proc_list pt             where pt.group_id = in_group_id               and pt.is_active = 'Y') loop    dbms_job.submit(v_job_id, z.proc_script);    insert into parallel_proc_log      (run_id, group_id, proc_script, job_id, start_time, end_time)    values      (v_run_id, in_group_id, z.proc_script, v_job_id, sysdate, null);    v_job_id_list     := v_job_id_list || ',' || to_char(v_job_id);    v_job_id_list_ext := v_job_id_list_ext || ' union all select ' ||                         to_char(v_job_id) || ' job_id from dual';  end loop;  commit;  v_job_id_list     := substr(v_job_id_list, 2);  v_job_id_list_ext := substr(v_job_id_list_ext, 12);  -- loop while not all jobs finished  loop    -- set parallel_proc_log.end_time for finished jobs    execute immediate 'update parallel_proc_log set end_time = sysdate where job_id in (' ||                      v_job_id_list_ext ||                      ' minus select job from user_jobs where job in (' ||                      v_job_id_list ||                      ') minus select job_id from parallel_proc_log where job_id in (' ||                      v_job_id_list || ') and end_time is not null)';    commit;    -- check whether all jobs finished    execute immediate 'select count(1) from user_jobs where job in (' ||                      v_job_id_list || ')'      into v_running_jobs_count;    -- if all jobs finished then exit    exit when v_running_jobs_count = 0;    -- sleep a little    sys.dbms_lock.sleep(0.1);  end loop;end;

create or replace procedure sleep(in_seconds integer) asbegin  sys.Dbms_Lock.Sleep(in_seconds);end;

insert into PARALLEL_PROC_GROUP_LIST(group_id, group_name) values(1, 'Тестовая группа');insert into PARALLEL_PROC_LIST(group_id, proc_script, is_active) values(1, 'begin sleep(5); end;', 'Y');insert into PARALLEL_PROC_LIST(group_id, proc_script, is_active) values(1, 'begin sleep(10); end;', 'Y');

begin  run_in_parallel(1);end;

select * from PARALLEL_PROC_LOG;

Протягивание остатков

create table ACCOUNT_BALANCE(  dt           DATE,  account_id   INTEGER,  balance_amt  NUMBER,  turnover_amt NUMBER);comment on column ACCOUNT_BALANCE.dt  is 'Дата и время остатка по счёту';comment on column ACCOUNT_BALANCE.account_id  is 'Номер счёта';comment on column ACCOUNT_BALANCE.balance_amt  is 'Остаток по счёту';comment on column ACCOUNT_BALANCE.turnover_amt  is 'Оборот по счёту';insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('01.01.2020 00:00:00','dd.mm.yyyy hh24.mi.ss'), 1, 23, 23);insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('05.01.2020 01:00:00','dd.mm.yyyy hh24.mi.ss'), 1, 45, 22);insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('05.01.2020 20:00:00','dd.mm.yyyy hh24.mi.ss'), 1, 44, -1);insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('05.01.2020 00:00:00','dd.mm.yyyy hh24.mi.ss'), 2, 67, 67);insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('05.01.2020 20:00:00','dd.mm.yyyy hh24.mi.ss'), 2, 77, 10);insert into account_balance(dt, account_id, balance_amt, turnover_amt) values(to_date('07.01.2020 00:00:00','dd.mm.yyyy hh24.mi.ss'), 2, 72, -5);

with cld as (select /*+ materialize*/   to_date('01.01.2020', 'dd.mm.yyyy') + level - 1 dt    from dual  connect by level <= 10),ab as (select trunc(dt) dt,         account_id,         max(balance_amt) keep(dense_rank last order by dt) balance_amt,         sum(turnover_amt) turnover_amt    from account_balance   group by trunc(dt), account_id),a as (select min(dt) min_dt, account_id from ab group by account_id),pre as (select cld.dt, a.account_id from cld left join a on cld.dt >= a.min_dt)select pre.dt,       pre.account_id,       last_value(ab.balance_amt ignore nulls) over(partition by pre.account_id order by pre.dt) balance_amt,       nvl(ab.turnover_amt, 0) turnover_amt  from pre  left join ab    on pre.dt = ab.dt   and pre.account_id = ab.account_id order by 2, 1;

Объединение нескольких историй в одну

create table HIST1(  primary_key_id INTEGER,  start_dt       DATE,  attribute1     NUMBER);insert into HIST1(primary_key_id, start_dt, attribute1) values(1, to_date('2014-01-01','yyyy-mm-dd'), 7);insert into HIST1(primary_key_id, start_dt, attribute1) values(1, to_date('2015-01-01','yyyy-mm-dd'), 8);insert into HIST1(primary_key_id, start_dt, attribute1) values(1, to_date('2016-01-01','yyyy-mm-dd'), 9);insert into HIST1(primary_key_id, start_dt, attribute1) values(2, to_date('2014-01-01','yyyy-mm-dd'), 17);insert into HIST1(primary_key_id, start_dt, attribute1) values(2, to_date('2015-01-01','yyyy-mm-dd'), 18);insert into HIST1(primary_key_id, start_dt, attribute1) values(2, to_date('2016-01-01','yyyy-mm-dd'), 19);create table HIST2(  primary_key_id INTEGER,  start_dt       DATE,  attribute2     NUMBER); insert into HIST2(primary_key_id, start_dt, attribute2) values(1, to_date('2015-01-01','yyyy-mm-dd'), 4);insert into HIST2(primary_key_id, start_dt, attribute2) values(1, to_date('2016-01-01','yyyy-mm-dd'), 5);insert into HIST2(primary_key_id, start_dt, attribute2) values(1, to_date('2017-01-01','yyyy-mm-dd'), 6);insert into HIST2(primary_key_id, start_dt, attribute2) values(2, to_date('2015-01-01','yyyy-mm-dd'), 14);insert into HIST2(primary_key_id, start_dt, attribute2) values(2, to_date('2016-01-01','yyyy-mm-dd'), 15);insert into HIST2(primary_key_id, start_dt, attribute2) values(2, to_date('2017-01-01','yyyy-mm-dd'), 16);create table HIST3(  primary_key_id INTEGER,  start_dt       DATE,  attribute3     NUMBER); insert into HIST3(primary_key_id, start_dt, attribute3) values(1, to_date('2016-01-01','yyyy-mm-dd'), 10);insert into HIST3(primary_key_id, start_dt, attribute3) values(1, to_date('2017-01-01','yyyy-mm-dd'), 20);insert into HIST3(primary_key_id, start_dt, attribute3) values(1, to_date('2018-01-01','yyyy-mm-dd'), 30);insert into HIST3(primary_key_id, start_dt, attribute3) values(2, to_date('2016-01-01','yyyy-mm-dd'), 110);insert into HIST3(primary_key_id, start_dt, attribute3) values(2, to_date('2017-01-01','yyyy-mm-dd'), 120);insert into HIST3(primary_key_id, start_dt, attribute3) values(2, to_date('2018-01-01','yyyy-mm-dd'), 130);

select primary_key_id,       start_dt,       nvl(lead(start_dt - 1)           over(partition by primary_key_id order by start_dt),           to_date('9999-12-31', 'yyyy-mm-dd')) as end_dt,       last_value(attribute1 ignore nulls) over(partition by primary_key_id order by start_dt) as attribute1,       last_value(attribute2 ignore nulls) over(partition by primary_key_id order by start_dt) as attribute2,       last_value(attribute3 ignore nulls) over(partition by primary_key_id order by start_dt) as attribute3  from (select primary_key_id,               start_dt,               max(attribute1) as attribute1,               max(attribute2) as attribute2,               max(attribute3) as attribute3          from (select primary_key_id,                       start_dt,                       attribute1,                       cast(null as number) attribute2,                       cast(null as number) attribute3                  from hist1                union all                select primary_key_id,                       start_dt,                       cast(null as number) attribute1,                       attribute2,                       cast(null as number) attribute3                  from hist2                union all                select primary_key_id,                       start_dt,                       cast(null as number) attribute1,                       cast(null as number) attribute2,                       attribute3                  from hist3) q1         group by primary_key_id, start_dt) q2 order by primary_key_id, start_dt;

Нормалайзер

create table DENORMALIZED_TABLE(  id  INTEGER,  val VARCHAR2(4000));insert into DENORMALIZED_TABLE(id, val) values(1, 'aaa,cccc,bb');insert into DENORMALIZED_TABLE(id, val) values(2, 'ddd');insert into DENORMALIZED_TABLE(id, val) values(3, 'fffff,e');

select id, regexp_substr(val, '[^,]+', 1, column_value) val, column_value  from denormalized_table,       table(cast(multiset                  (select level                     from dual                   connect by regexp_instr(val, '[^,]+', 1, level) > 0) as                  sys.odcinumberlist)) order by id, column_value;

Визуализация в формате SVG

create table graph_data(dt date, val number, radius number);insert into graph_data(dt, val, radius) values (to_date('01.01.2020','dd.mm.yyyy'), 12, 3);insert into graph_data(dt, val, radius) values (to_date('02.01.2020','dd.mm.yyyy'), 15, 4);insert into graph_data(dt, val, radius) values (to_date('05.01.2020','dd.mm.yyyy'), 17, 5);insert into graph_data(dt, val, radius) values (to_date('06.01.2020','dd.mm.yyyy'), 13, 6);insert into graph_data(dt, val, radius) values (to_date('08.01.2020','dd.mm.yyyy'),  3, 7);insert into graph_data(dt, val, radius) values (to_date('10.01.2020','dd.mm.yyyy'), 20, 8);insert into graph_data(dt, val, radius) values (to_date('11.01.2020','dd.mm.yyyy'), 18, 9);

<line x1="94" x2="94" y1="15" y2="675" style="stroke:rgb(150,255,255); stroke-width:1px"/><circle cx="30" cy="279" r="3" style="fill:rgb(255,0,0)"/><text x="7" y="688" font-size="10" fill="rgb(0,150,255)">2020-01-01</text>

with const as (select 700 viewbox_width,         700 viewbox_height,         30 left_margin,         30 right_margin,         15 top_margin,         25 bottom_margin,         max(dt) - min(dt) + 1 num_vertical_lines,         11 num_horizontal_lines,         'rgb(150,255,255)' stroke_vertical_lines,         '1px' stroke_width_vertical_lines,         10 font_size_dates,         'rgb(0,150,255)' fill_dates,         23 x_dates_pad,         13 y_dates_pad,         'rgb(150,255,255)' stroke_horizontal_lines,         '1px' stroke_width_horizontal_lines,         10 font_size_values,         'rgb(0,150,255)' fill_values,         4 x_values_pad,         2 y_values_pad,         'rgb(255,0,0)' fill_circles,         'rgb(51,102,0)' stroke_graph,         '1px' stroke_width_graph,         min(dt) min_dt,         max(dt) max_dt,         max(val) max_val    from graph_data),gd1 as (select graph_data.dt,         const.left_margin +         (const.viewbox_width - const.left_margin - const.right_margin) *         (graph_data.dt - const.min_dt) / (const.max_dt - const.min_dt) x,         const.viewbox_height - const.bottom_margin -         (const.viewbox_height - const.top_margin - const.bottom_margin) *         graph_data.val / const.max_val y,         graph_data.radius    from graph_data, const),gd2 as (select dt,         round(nvl(lag(x) over(order by dt), x)) prev_x,         round(x) x,         round(nvl(lag(y) over(order by dt), y)) prev_y,         round(y) y,         radius    from gd1)/* header */select '<?xml version="1.0" encoding="UTF-8" standalone="no"?>' txt  from dualunion allselect '<svg version="1.1" width="' || viewbox_width || '" height="' ||       viewbox_height || '" viewBox="0 0 ' || viewbox_width || ' ' ||       viewbox_height ||       '" style="background:yellow" baseProfile="full" xmlns="http://personeltest.ru/away/www.w3.org/2000/svg" xmlns:xlink="http://personeltest.ru/away/www.w3.org/1999/xlink" xmlns:ev="http://personeltest.ru/away/www.w3.org/2001/xml-events">'  from constunion allselect '<title>Test graph</title>'  from dualunion allselect '<desc>Test graph</desc>'  from dualunion allselect '<rect width="' || viewbox_width || '" height="' || viewbox_height ||       '" style="fill:white" />'  from constunion all/* vertical lines */select '<line x1="' ||       to_char(round(left_margin +                     (viewbox_width - left_margin - right_margin) *                     (level - 1) / (num_vertical_lines - 1))) || '" x2="' ||       to_char(round(left_margin +                     (viewbox_width - left_margin - right_margin) *                     (level - 1) / (num_vertical_lines - 1))) || '" y1="' ||       to_char(round(top_margin)) || '" y2="' ||       to_char(round(viewbox_height - bottom_margin)) || '" style="stroke:' ||       const.stroke_vertical_lines || '; stroke-width:' ||       const.stroke_width_vertical_lines || '"/>'  from constconnect by level <= num_vertical_linesunion all/* dates under vertical lines */select '<text x="' ||       to_char(round(left_margin +                     (viewbox_width - left_margin - right_margin) *                     (level - 1) / (num_vertical_lines - 1) - x_dates_pad)) ||       '" y="' ||       to_char(round(viewbox_height - bottom_margin + y_dates_pad)) ||       '" font-size="' || font_size_dates || '" fill="' || fill_dates || '">' ||       to_char(min_dt + level - 1, 'yyyy-mm-dd') || '</text>'  from constconnect by level <= num_vertical_linesunion all/* horizontal lines */select '<line x1="' || to_char(round(left_margin)) || '" x2="' ||       to_char(round(viewbox_width - right_margin)) || '" y1="' ||       to_char(round(top_margin +                     (viewbox_height - top_margin - bottom_margin) *                     (level - 1) / (num_horizontal_lines - 1))) || '" y2="' ||       to_char(round(top_margin +                     (viewbox_height - top_margin - bottom_margin) *                     (level - 1) / (num_horizontal_lines - 1))) ||       '" style="stroke:' || const.stroke_horizontal_lines ||       '; stroke-width:' || const.stroke_width_horizontal_lines || '"/>'  from constconnect by level <= num_horizontal_linesunion all/* values near horizontal lines */select '<text text-anchor="end" x="' ||       to_char(round(left_margin - x_values_pad)) || '" y="' ||       to_char(round(viewbox_height - bottom_margin -                     (viewbox_height - top_margin - bottom_margin) *                     (level - 1) / (num_horizontal_lines - 1) +                     y_values_pad)) || '" font-size="' || font_size_values ||       '" fill="' || fill_values || '">' ||       to_char(round(max_val / (num_horizontal_lines - 1) * (level - 1), 2)) ||       '</text>'  from constconnect by level <= num_horizontal_linesunion all/* circles */select '<circle cx="' || to_char(gd2.x) || '" cy="' || to_char(gd2.y) ||       '" r="' || gd2.radius || '" style="fill:' || const.fill_circles ||       '"/>'  from gd2, constunion all/* graph data */select '<line x1="' || to_char(gd2.prev_x) || '" x2="' || to_char(gd2.x) ||       '" y1="' || to_char(gd2.prev_y) || '" y2="' || to_char(gd2.y) ||       '" style="stroke:' || const.stroke_graph || '; stroke-width:' ||       const.stroke_width_graph || '"/>'  from gd2, constunion all/* footer */select '</svg>' from dual;

Приложение поиска по метаданным Oracle

• Где в коде Oracle зашита константа 61209, представляющая собой номер счёта второго порядка?
• Где в коде и на каких серверах используется таблица accounts (в т.ч. через database link)?
• С какого сервера из какой хранимой процедуры или триггера приходит сгенерированная программистом ошибка, например, ORA-20001 Курс валюты не найден?
• Прописан ли планируемый к удалению индекс IX_CLIENTID где-либо явным образом в хинтах оптимизатора в SQL-запросах?
• Используются ли где-либо (в т.ч. через database link) планируемые к удалению таблица, поле, процедура, функция и т.д.?
• Где в коде явно зашит чей-то е-мэйл или номер телефона? Такие вещи лучше выносить из серверных объектов в настроечные таблицы.
• Где в коде на серверах используется зависящий от версии Oracle функционал? Например, функция wm_concat выдаёт различный тип данных на выходе в зависимости от версии Oracle. Это может быть критично и требует внимания при миграции на более новую версию.
• Где в коде используется какой-либо редкий приём, на который программисту хочется посмотреть, как на образец? Например, поискать в коде Oracle примеры использования функций sys_connect_by_path, regexp_instr или хинта push_subq.

Автор: Михаил Гричик, эксперт профессионального сообщества Сбербанка SberProfi DWH/BigData.

Профессиональное сообщество SberProfi DWH/BigData отвечает за развитие компетенций в таких направлениях, как экосистема Hadoop, Teradata, Oracle DB, GreenPlum, а также BI инструментах Qlik, SAP BO, Tableau и др.

• не безразлично, сколько у вас накопилось имущества и как вести его учёт
• интересно, как заставить своё имущество приносить вам дополнительный доход
• хочется понять, какой же из способов вложения денег лучший: недвижимость, депозиты или акции
• любопытно, что посоветует по этому вопросу анализ данных Сбербанка

• учёт и оценка всего имеющегося в наличии имущества
• учёт доходов и расходов, а также разницы между доходами и расходами, т.е. динамики накопления имущества

Учёт и оценка всего имеющегося в наличии имущества

• недвижимость
• земельные участки
• транспортные средства
• вклады в банках
• кредитные обязательства (с минусом)
• инвестиции (акции, облигации, ...)
• стоимость собственного бизнеса
• прочее имущество

Учёт доходов, расходов и динамики накопления имущества

• динамика стоимости квадратного метра в Москве
• база предложений по продаже и аренде недвижимости в Москве и ближнем Подмосковье
• динамика средней годовой процентной ставки по депозитам
• динамика уровня рублёвой инфляции
• динамика индекса Мосбиржи полной доходности брутто (MCFTR)
• котировки акций Мосбиржи и данные о выплаченных дивидендах

Исторический анализ рынка недвижимости

Исторический анализ ставок по депозитам

• Можно фиксировать ставку по пополняемым вкладам в благоприятное время на несколько месяцев вперёд
• Ежемесячная капитализация, свойственная многим учтённым в этих усреднённых данных вкладам, за счёт сложных процентов добавляет прибыли
• Выше были учтены ставки по топ-10 банкам по информации от Банка России, вне топ-10 можно найти ставки несколько выше

Исторический анализ рынка акций

Автор: Михаил Гричик, эксперт профессионального сообщества Сбербанка SberProfi DWH/BigData.

Профессиональное сообщество SberProfi DWH/BigData отвечает за развитие компетенций в таких направлениях, как экосистема Hadoop, Teradata, Oracle DB, GreenPlum, а также BI инструментах Qlik, SAP BO, Tableau и др.

1. DWH: ода Вертике

• 2*CPU по 20 ядер
• 256Gb RAM
• 2*10G сеть
• быстрые 10k SAS HDD RAID 10 (в 2017/18, когда мы планировали обновление и обновляли RAID массивы, SSD стоили как чугунный мост и были не такими надежными как сейчас)

2. ETL

• до появления Loader сотни Гб бинарников CDR парсились джобами Talend. Loader и появился из доработки джобов Talend, которые перестали справляться с нагрузкой;
• внутренняя команда иногда переписывает за подрядчиком пайплайны, созданные в их Loader, и время на обработку данных уменьшается. Понятно, что ситуация разовая, и 1 Tb в сутки из бинарников вряд ли Talend распарсит, но факт есть факт.

3. Визуализация данных

Есть у этих самых аналитиков любовь к типам диаграмм 'водопад':

Прошу прощения за низкое качество изображения, но суть передана верно

4. Data science (machine learning)

5. Agile? Нет У нас своё

• Быстроменяющаяся архитектура BI-решения является черным ящиком для всех смежных команд. Эта статья первый из шагов по расшариванию знаний о нашем решении.
• Отток в команде потенциально критичен. Однако стабильность команды и преемственность нивелируют этот минус.

• Высокий TTM и высокая пропускная способность команды в целом. Весь проектный портфель компании (почти во всех проектах есть фичи на отдел) составляет 15-20% общего объема разработки отдела. Остальное прямые пожелания конечных бизнес-заказчиков, реализуемые с минимумом бюрократии.
• Стабильно высокий CSI, демонстрирующий правильность выбранного подхода в организации разработки. Один раз в квартал мы проводим опрос среди бизнес-заказчиков. В 4Q20 из 43 респондентов ответили 21. По итогу получили 4,89 из 5. Это упавший CSI, хотя я предполагал падение до 4,5. Стандартно у нас ближе к 5. Объясняется это гибкостью в подходе к реализациям задумок бизнес-заказчиков и скоростью появления конечного результата с максимально эффективным использованием имеющихся инструментов/технологий.

В опросе CSI также можно оставить комментарий, например такой

6. Заключение

О чем будем говорить

Автоматизированное тестирование методом черного ящика в хранилище данных

Перевод банковских процессов на ETL-as-Service при помощи Informatica Power Center

Привет, я Дмитрий Логвиненко Data Engineer отдела аналитики группы компаний Везёт.

Я расскажу вам о замечательном инструменте для разработки ETL-процессов Apache Airflow. Но Airflow настолько универсален и многогранен, что вам стоит присмотреться к нему даже если вы не занимаетесь потоками данных, а имеете потребность периодически запускать какие-либо процессы и следить за их выполнением.

И да, я буду не только рассказывать, но и показывать: в программе много кода, скриншотов и рекомендаций.

Что обычно видишь, когда гуглишь слово Airflow / Wikimedia Commons

Введение

Apache Airflow он прямо как Django:

• написан на Python,
• есть отличная админка,
• неограниченно расширяем,

только лучше, да и сделан совсем для других целей, а именно (как написано до ката):

• запуск и мониторинг задач на неограниченном количестве машин (сколько вам позволит Celery/Kubernetes и ваша совесть)
• с динамической генерацией workflow из очень легкого для написания и восприятия Python-кода
• и возможностью связывать друг с друг любые базы данных и API с помощью как готовых компонентов, так и самодельных плагинов (что делается чрезвычайно просто).

Мы используем Apache Airflow так:

• собираем данные из различных источников (множество инстансов SQL Server и PostgreSQL, различные API с метриками приложений, даже 1С) в DWH и ODS (у нас это Vertica и Clickhouse).
• как продвинутый cron, который запускает процессы консолидации данных на ODS, а также следит за их обслуживанием.

До недавнего времени наши потребности покрывал один небольшой сервер на 32 ядрах и 50 GB оперативки. В Airflow при этом работает:

• более 200 дагов (собственно workflows, в которые мы набили задачки),
• в каждом в среднем по 70 тасков,
• запускается это добро (тоже в среднем) раз в час.

А о том, как мы расширялись, я напишу ниже, а сейчас давайте определим ber-задачу, которую мы будем решать:

Есть три исходных SQL Serverа, на каждом по 50 баз данных инстансов одного проекта, соответственно, структура у них одинаковая (почти везде, муа-ха-ха), а значит в каждой есть таблица Orders (благо таблицу с таким названием можно затолкать в любой бизнес). Мы забираем данные, добавляя служебные поля (сервер-источник, база-источник, идентификатор ETL-задачи) и наивным образом бросим их в, скажем, Vertica.

Поехали!

Часть основная, практическая (и немного теоретическая)

Зачем оно нам (и вам)

Когда деревья были большими, а я был простым SQL-щиком в одном российском ритейле, мы шпарили ETL-процессы aka потоки данных с помощью двух доступных нам средств:

• Informatica Power Center крайне развесистая система, чрезвычайно производительная, со своими железками, собственным версионированием. Использовал я дай бог 1% её возможностей. Почему? Ну, во-первых, этот интерфейс где-то из нулевых психически давил на нас. Во-вторых, эта штуковина заточена под чрезвычайно навороченные процессы, яростное переиспользование компонентов и другие очень-важные-энтерпрайз-фишечки. Про то что стоит она, как крыло Airbus A380/год, мы промолчим.

  
  
  Осторожно, скриншот может сделать людям младше 30 немного больно

  

  
  
  
  

• SQL Server Integration Server этим товарищем мы пользовались в своих внутрипроектных потоках. Ну а в самом деле: SQL Server мы уже используем, и не юзать его ETL-тулзы было бы как-то неразумно. Всё в нём в хорошо: и интерфейс красивый, и отчётики выполнения Но не за это мы любим программные продукты, ох не за это. Версионировать его dtsx (который представляет собой XML с перемешивающимися при сохранении нодами) мы можем, а толку? А сделать пакет тасков, который перетащит сотню таблиц с одного сервера на другой? Да что сотню, у вас от двадцати штук отвалится указательный палец, щёлкающий по мышиной кнопке. Но выглядит он, определенно, более модно:

  
  

  

  
  

Мы безусловно искали выходы. Дело даже почти дошло до самописного генератора SSIS-пакетов...

а потом меня нашла новая работа. А на ней меня настиг Apache Airflow.

Когда я узнал, что описания ETL-процессов это простой Python-код, я только что не плясал от радости. Вот так потоки данных подверглись версионированию и диффу, а ссыпать таблицы с единой структурой из сотни баз данных в один таргет стало делом Python-кода в полтора-два 13 экрана.

Собираем кластер

Давайте не устраивать совсем уж детский сад, и не говорить тут о совершенно очевидных вещах, вроде установки Airflow, выбранной вами БД, Celery и других дел, описанных в доках.

Чтобы мы могли сразу приступить к экспериментам, я набросал docker-compose.yml в котором:

• Поднимем собственно Airflow: Scheduler, Webserver. Там же будет крутится Flower для мониторинга Celery-задач (потому что его уже затолкали в apache/airflow:1.10.10-python3.7, а мы и не против);
• PostgreSQL, в который Airflow будет писать свою служебную информацию (данные планировщика, статистика выполнения и т. д.), а Celery отмечать завершенные таски;
• Redis, который будет выступать брокером задач для Celery;
• Celery worker, который и займется непосредственным выполнением задачек.
• В папку ./dags мы будет складывать наши файлы с описанием дагов. Они будут подхватываться на лету, поэтому передёргивать весь стек после каждого чиха не нужно.

Кое-где код в примерах приведен не полностью (чтобы не загромождать текст), а где-то он модифицируется в процессе. Цельные работающие примеры кода можно посмотреть в репозитории https://github.com/dm-logv/airflow-tutorial.

version: '3.4'x-airflow-config: &airflow-config  AIRFLOW__CORE__DAGS_FOLDER: /dags  AIRFLOW__CORE__EXECUTOR: CeleryExecutor  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow  AIRFLOW__CORE__PARALLELISM: 128  AIRFLOW__CORE__DAG_CONCURRENCY: 16  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflowx-airflow-base: &airflow-base  image: apache/airflow:1.10.10-python3.7  entrypoint: /bin/bash  restart: always  volumes:    - ./dags:/dags    - ./requirements.txt:/requirements.txtservices:  # Redis as a Celery broker  broker:    image: redis:6.0.5-alpine  # DB for the Airflow metadata  airflow-db:    image: postgres:10.13-alpine    environment:      - POSTGRES_USER=airflow      - POSTGRES_PASSWORD=airflow      - POSTGRES_DB=airflow    volumes:      - ./db:/var/lib/postgresql/data  # Main container with Airflow Webserver, Scheduler, Celery Flower  airflow:    <<: *airflow-base    environment:      <<: *airflow-config      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'      AIRFLOW__SCHEDULER__MAX_THREADS: 8      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10    depends_on:      - airflow-db      - broker    command: >      -c " sleep 10 &&           pip install --user -r /requirements.txt &&           /entrypoint initdb &&          (/entrypoint webserver &) &&          (/entrypoint flower &) &&           /entrypoint scheduler"    ports:      # Celery Flower      - 5555:5555      # Airflow Webserver      - 8080:8080  # Celery worker, will be scaled using `--scale=n`  worker:    <<: *airflow-base    environment:      <<: *airflow-config    command: >      -c " sleep 10 &&           pip install --user -r /requirements.txt &&           /entrypoint worker"    depends_on:      - airflow      - airflow-db      - broker

Примечания:

• В сборке композа я во многом опирался на известный образ puckel/docker-airflow обязательно посмотрите. Может, вам в жизни больше ничего и не понадобится.
• Все настройки Airflow доступны не только через airflow.cfg, но и через переменные среды (слава разработчикам), чем я злостно воспользовался.
• Естественно, он не production-ready: я намеренно не ставил heartbeats на контейнеры, не заморачивался с безопасностью. Но минимум, подходящий для наших экспериментиков я сделал.
• Обратите внимание, что:
  
  • Папка с дагами должна быть доступна как планировщику, так и воркерам.
  • То же самое касается и всех сторонних библиотек они все должны быть установлены на машины с шедулером и воркерами.

Ну а теперь просто:

$ docker-compose up --scale worker=3

После того, как всё поднимется, можно смотреть на веб-интерфейсы:

• Airflow: http://127.0.0.1:8080/admin/
• Flower: http://127.0.0.1:5555/dashboard

Основные понятия

Если вы ничего не поняли во всех этих дагах, то вот краткий словарик:

• Scheduler самый главный дядька в Airflow, контролирующий, чтобы вкалывали роботы, а не человек: следит за расписанием, обновляет даги, запускает таски.

  
  

  Вообще, в старых версиях, у него были проблемы с памятью (нет, не амнезия, а утечки) и в конфигах даже остался легаси-параметр run_duration интервал его перезапуска. Но сейчас всё хорошо.

  
  

• DAG (он же даг) направленный ацикличный граф, но такое определение мало кому что скажет, а по сути это контейнер для взаимодействующих друг с другом тасков (см. ниже) или аналог Package в SSIS и Workflow в Informatica.

  
  

  Помимо дагов еще могут быть сабдаги, но мы до них скорее всего не доберёмся.

  
  

• DAG Run инициализированный даг, которому присвоен свой execution_date. Даграны одного дага могут вполне работать параллельно (если вы, конечно, сделали свои таски идемпотентными).

  
  

• Operator это кусочки кода, ответственные за выполнение какого-либо конкретного действия. Есть три типа операторов:

  
  
  • action, как например наш любимый PythonOperator, который в силах выполнить любой (валидный) Python-код;
  • transfer, которые перевозят данные с места на место, скажем, MsSqlToHiveTransfer;
  • sensor же позволит реагировать или притормозить дальнейшее выполнение дага до наступления какого-либо события. HttpSensor может дергать указанный эндпойнт, и когда дождется нужный ответ, запустить трансфер GoogleCloudStorageToS3Operator. Пытливый ум спросит: зачем? Ведь можно делать повторы прямо в операторе! А затем, чтобы не забивать пул тасков подвисшими операторами. Сенсор запускается, проверяет и умирает до следующей попытки.
  
  

• Task объявленные операторы вне зависимости от типа и прикрепленные к дагу повышаются до чина таска.

  
  

• Task instance когда генерал-планировщик решил, что таски пора отправлять в бой на исполнители-воркеры (прямо на месте, если мы используем LocalExecutor или на удалённую ноду в случае с CeleryExecutor), он назначает им контекст (т. е. комплект переменных параметров выполнения), разворачивает шаблоны команд или запросов и складывает их в пул.

  
  

Генерируем таски

Сперва обозначим общую схему нашего дага, а затем будем всё больше и больше погружаться в детали, потому что мы применяем некоторые нетривиальные решения.

Итак, в простейшем виде подобный даг будет выглядеть так:

from datetime import timedelta, datetimefrom airflow import DAGfrom airflow.operators.python_operator import PythonOperatorfrom commons.datasources import sql_server_dsdag = DAG('orders',          schedule_interval=timedelta(hours=6),          start_date=datetime(2020, 7, 8, 0))def workflow(**context):    print(context)for conn_id, schema in sql_server_ds:    PythonOperator(        task_id=schema,        python_callable=workflow,        provide_context=True,        dag=dag)

Давайте разбираться:

• Сперва импортируем нужные либы и кое что ещё;
• sql_server_ds это List[namedtuple[str, str]] с именами коннектов из Airflow Connections и базами данных из которых мы будем забирать нашу табличку;
• dag объявление нашего дага, которое обязательно должно лежать в globals(), иначе Airflow его не найдет. Дагу также нужно сказать:
  
  • что его зовут orders это имя потом будет маячить в веб-интерфейсе,
  • что работать он будет, начиная с полуночи восьмого июля,
  • а запускать он должен, примерно каждые 6 часов (для крутых парней здесь вместо timedelta() допустима cron-строка 0 0 0/6 ? * * *, для менее крутых выражение вроде @daily);
• workflow() будет делать основную работу, но не сейчас. Сейчас мы просто высыпем наш контекст в лог.
• А теперь простая магия создания тасков:
  
  • пробегаем по нашим источникам;
  • инициализируем PythonOperator, который будет выполнять нашу пустышку workflow(). Не забывайте указывать уникальное (в рамках дага) имя таска и подвязывать сам даг. Флаг provide_context в свою очередь насыпет в функцию дополнительных аргументов, которые мы бережно соберём с помощью **context.

Пока на этом всё. Что мы получили:

• новый даг в веб-интерфейсе,
• полторы сотни тасков, которые будут выполняться параллельно (если то позволят настройки Airflow, Celery и мощности серверов).

Ну, почти получили.

Зависимости кто будет ставить?

Чтобы всё это дело упростить я вкорячил в docker-compose.yml обработку requirements.txt на всех нодах.

Вот теперь понеслась:

Серые квадратики task instances, обработанные планировщиком.

Немного ждем, задачи расхватывают воркеры:

Зеленые, понятное дело, успешно отработавшие. Красные не очень успешно.

Кстати, на нашем проде никакой папки ./dags, синхронизирующейся между машинами нет всё даги лежат в git на нашем Gitlab, а Gitlab CI раскладывает обновления на машины при мёрдже в master.

Немного о Flower

Пока воркеры молотят наши тасочки-пустышки, вспомним про еще один инструмент, который может нам кое-что показать Flower.

Самая первая страничка с суммарной информацией по нодам-воркерам:

Самая насыщенная страничка с задачами, отправившимися в работу:

Самая скучная страничка с состоянием нашего брокера:

Самая яркая страничка с графиками состояния тасков и их временем выполнения:

Догружаем недогруженное

Итак, все таски отработали, можно уносить раненых.

А раненых оказалось немало по тем или иным причинами. В случае правильного использования Airflow вот эти самые квадраты говорят о том, что данные определенно не доехали.

Нужно смотреть лог и перезапускать упавшие task instances.

Жмякнув на любой квадрат, увидим доступные нам действия:

Можно взять, и сделать Clear упавшему. То есть, мы забываем о том, что там что-то завалилось, и тот же самый инстанс таска уйдет планировщику.

Понятно, что делать так мышкой со всеми красными квадратами не очень гуманно не этого мы ждем от Airflow. Естественно, у нас есть оружие массового поражения: Browse/Task Instances

Выберем всё разом и обнулим нажмем правильный пункт:

После очистки наши такси выглядят так (они уже ждут не дождутся, когда шедулер их запланирует):

Соединения, хуки и прочие переменные

Самое время посмотреть на следующий DAG, update_reports.py:

from collections import namedtuplefrom datetime import datetime, timedeltafrom textwrap import dedentfrom airflow import DAGfrom airflow.contrib.operators.vertica_operator import VerticaOperatorfrom airflow.operators.email_operator import EmailOperatorfrom airflow.utils.trigger_rule import TriggerRulefrom commons.operators import TelegramBotSendMessagedag = DAG('update_reports',          start_date=datetime(2020, 6, 7, 6),          schedule_interval=timedelta(days=1),          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})Report = namedtuple('Report', 'source target')reports = [Report(f'{table}_view', table) for table in [    'reports.city_orders',    'reports.client_calls',    'reports.client_rates',    'reports.daily_orders',    'reports.order_duration']]email = EmailOperator(    task_id='email_success', dag=dag,    to='{{ var.value.all_the_kings_men }}',    subject='DWH Reports updated',    html_content=dedent("""Господа хорошие, отчеты обновлены"""),    trigger_rule=TriggerRule.ALL_SUCCESS)tg = TelegramBotSendMessage(    task_id='telegram_fail', dag=dag,    tg_bot_conn_id='tg_main',    chat_id='{{ var.value.failures_chat }}',    message=dedent("""\         Наташ, просыпайся, мы {{ dag.dag_id }} уронили        """),    trigger_rule=TriggerRule.ONE_FAILED)for source, target in reports:    queries = [f"TRUNCATE TABLE {target}",               f"INSERT INTO {target} SELECT * FROM {source}"]    report_update = VerticaOperator(        task_id=target.replace('reports.', ''),        sql=queries, vertica_conn_id='dwh',        task_concurrency=1, dag=dag)    report_update >> [email, tg]

Все ведь когда-нибудь делали обновлялку отчетов? Это снова она: есть список источников, откуда забрать данные; есть список, куда положить; не забываем посигналить, когда всё случилось или сломалось (ну это не про нас, нет).

Давайте снова пройдемся по файлу и посмотрим на новые непонятные штуки:

• from commons.operators import TelegramBotSendMessage нам ничто не мешает делать свои операторы, чем мы и воспользовались, сделав небольшую обёрточку для отправки сообщений в Разблокированный. (Об этом операторе мы еще поговорим ниже);
• default_args={} даг может раздавать одни и те же аргументы всем своим операторам;
• to='{{ var.value.all_the_kings_men }}' поле to у нас будет не захардкоженным, а формируемым динамически с помощью Jinja и переменной со списком email-ов, которую я заботливо положил в Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS условие запуска оператора. В нашем случае, письмо полетит боссам только если все зависимости отработали успешно;
• tg_bot_conn_id='tg_main' аргументы conn_id принимают в себя идентификаторы соединений, которые мы создаем в Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED сообщения в Telegram улетят только при наличии упавших тасков;
• task_concurrency=1 запрещаем одновременный запуск нескольких task instances одного таска. В противном случае, мы получим одновременный запуск нескольких VerticaOperator (смотрящих на одну таблицу);

• report_update >> [email, tg] все VerticaOperator сойдутся в отправке письма и сообщения, вот так:
  

  
  

  Но так как у операторов-нотификаторов стоят разные условия запуска, работать будет только один. В Tree View всё выглядит несколько менее наглядно:
  

  
  

Скажу пару слов о макросах и их друзьях переменных.

Макросы это Jinja-плейсхолдеры, которые могут подставлять разную полезную информацию в аргументы операторов. Например, так:

SELECT    id,    payment_dtm,    payment_type,    client_idFROM orders.paymentsWHERE    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} развернется в содержимое переменной контекста execution_date в формате YYYY-MM-DD: 2020-07-14. Самое приятное, что переменные контекста прибиваются гвоздями к определенному инстансу таска (квадратику в Tree View), и при перезапуске плейсхолдеры раскроются в те же самые значения.

Присвоенные значения можно смотреть с помощью кнопки Rendered на каждом таск-инстансе. Вот так у таска с отправкой письма:

А так у таски с отправкой сообщения:

Полный список встроенных макросов для последней доступной версии доступен здесь: Macros Reference

Более того, с помощью плагинов, мы можем объявлять собственные макросы, но это уже совсем другая история.

Помимо предопределенных штук, мы можем подставлять значения своих переменных (выше в коде я уже этим воспользовался). Создадим в Admin/Variables пару штук:

Всё, можно пользоваться:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

В значении может быть скаляр, а может лежать и JSON. В случае JSON-а:

bot_config{    "bot": {        "token": 881hskdfASDA16641,        "name": "Verter"    },    "service": "TG"}

просто используем путь к нужному ключу: {{ var.json.bot_config.bot.token }}.

Скажу буквально одно слово и покажу один скриншот про соединения. Тут всё элементарно: на странице Admin/Connections создаем соединение, складываем туда наши логины/пароли и более специфичные параметры. Вот так:

Пароли можно шифровать (более тщательно, чем в варианте по умолчанию), а можно не указывать тип соединения (как я сделал для tg_main) дело в том, что список типов зашит в моделях Airflow и расширению без влезания в исходники не поддается (если вдруг я чего-то не догуглил прошу меня поправить), но получить креды просто по имени нам ничто не помешает.

А еще можно сделать несколько соединений с одним именем: в таком случае метод BaseHook.get_connection(), который достает нам соединения по имени, будет отдавать случайного из нескольких тёзок (было бы логичнее сделать Round Robin, но оставим это на совести разработчиков Airflow).

Variables и Connections, безусловно, классные средства, но важно не потерять баланс: какие части ваших потоков вы храните собственно в коде, а какие отдаете на хранение Airflow. C одной стороны быстро поменять значение, например, ящик рассылки, может быть удобно через UI. А с другой это всё-таки возврат к мышеклику, от которого мы (я) хотели избавиться.

Работа с соединениями это одна из задач хуков. Вообще хуки Airflow это точки подключения его к сторонним сервисам и библиотекам. К примеру, JiraHook откроет для нас клиент для взаимодействия с Jira (можно задачки подвигать туда-сюда), а с помощью SambaHook можно запушить локальный файл на smb-точку.

Разбираем кастомный оператор

И мы вплотную подобрались к тому, чтобы посмотреть на то, как сделан TelegramBotSendMessage

Код commons/operators.py с собственно оператором:

from typing import Unionfrom airflow.operators import BaseOperatorfrom commons.hooks import TelegramBotHook, TelegramBotclass TelegramBotSendMessage(BaseOperator):    """Send message to chat_id using TelegramBotHook    Example:        >>> TelegramBotSendMessage(        ...     task_id='telegram_fail', dag=dag,        ...     tg_bot_conn_id='tg_bot_default',        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',        ...     message='{{ dag.dag_id }} failed :(',        ...     trigger_rule=TriggerRule.ONE_FAILED)    """    template_fields = ['chat_id', 'message']    def __init__(self,                 chat_id: Union[int, str],                 message: str,                 tg_bot_conn_id: str = 'tg_bot_default',                 *args, **kwargs):        super().__init__(*args, **kwargs)        self._hook = TelegramBotHook(tg_bot_conn_id)        self.client: TelegramBot = self._hook.client        self.chat_id = chat_id        self.message = message    def execute(self, context):        print(f'Send "{self.message}" to the chat {self.chat_id}')        self.client.send_message(chat_id=self.chat_id,                                 message=self.message)

Здесь, как и остальное в Airflow, всё очень просто:

• Отнаследовались от BaseOperator, который реализует довольно много Airflow-специфичных штук (посмотрите на досуге)
• Объявили поля template_fields, в которых Jinja будет искать макросы для обработки.
• Организовали правильные аргументы для __init__(), расставили умолчания, где надо.
• Об инициализации предка тоже не забыли.
• Открыли соответствующий хук TelegramBotHook, получили от него объект-клиент.
• Оверрайднули (переопределили) метод BaseOperator.execute(), который Airfow будет подергивать, когда наступит время запускать оператор в нем мы и реализуем основное действие, на забыв залогироваться. (Логируемся, кстати, прямо в stdout и stderr Airflow всё перехватит, красиво обернет, разложит, куда надо.)

Давайте смотреть, что у нас в commons/hooks.py. Первая часть файлика, с самим хуком:

from typing import Unionfrom airflow.hooks.base_hook import BaseHookfrom requests_toolbelt.sessions import BaseUrlSessionclass TelegramBotHook(BaseHook):    """Telegram Bot API hook    Note: add a connection with empty Conn Type and don't forget    to fill Extra:        {"bot_token": "YOuRAwEsomeBOtToKen"}    """    def __init__(self,                 tg_bot_conn_id='tg_bot_default'):        super().__init__(tg_bot_conn_id)        self.tg_bot_conn_id = tg_bot_conn_id        self.tg_bot_token = None        self.client = None        self.get_conn()    def get_conn(self):        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson        self.tg_bot_token = extra['bot_token']        self.client = TelegramBot(self.tg_bot_token)        return self.client

Я даже не знаю, что тут можно объяснять, просто отмечу важные моменты:

• Наследуемся, думаем над аргументами в большинстве случаев он будет один: conn_id;
• Переопределяем стандартные методы: я ограничился get_conn(), в котором я получаю параметры соединения по имени и всего-навсего достаю секцию extra (это поле для JSON), в которую я (по своей же инструкции!) положил токен Telegram-бота: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• Создаю экземпляр нашего TelegramBot, отдавая ему уже конкретный токен.

Вот и всё. Получить клиент из хука можно c помощью TelegramBotHook().clent или TelegramBotHook().get_conn().

И вторая часть файлика, в котором я сделать микрообёрточку для Telegram REST API, чтобы не тащить тот же python-telegram-bot ради одного метода sendMessage.

class TelegramBot:    """Telegram Bot API wrapper    Examples:        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hi, darling')        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hi, darling', chat_id=-1762374628374)    """    API_ENDPOINT = 'https://api.telegram.org/bot{}/'    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)        self.session = BaseUrlSession(self._base_url)        self.chat_id = chat_id    def send_message(self, message: str, chat_id: Union[int, str] = None):        method = 'sendMessage'        payload = {'chat_id': chat_id or self.chat_id,                   'text': message,                   'parse_mode': 'MarkdownV2'}        response = self.session.post(method, data=payload).json()        if not response.get('ok'):            raise TelegramBotException(response)class TelegramBotException(Exception):    def __init__(self, *args, **kwargs):        super().__init__((args, kwargs))

Правильный путь сложить всё это: TelegramBotSendMessage, TelegramBotHook, TelegramBot в плагин, положить в общедоступный репозиторий, и отдать в Open Source.

Пока мы всё это изучали, наши обновления отчетов успели успешно завалиться и отправить мне в канал сообщение об ошибке. Пойду проверять, что опять не так...

В нашем даге что-то сломалось! А ни этого ли мы ждали? Именно!

Наливать-то будешь?

Чувствуете, что-то я пропустил? Вроде бы обещал данные из SQL Server в Vertica переливать, и тут взял и съехал с темы, негодяй!

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

План у нас был такой:

1. Сделать даг
2. Нагенерить таски
3. Посмотреть, как всё красиво
4. Присваивать заливкам номера сессий
5. Забрать данные из SQL Server
6. Положить данные в Vertica
7. Собрать статистику

Итак, чтобы всё это запустить, я сделал маленькое дополнение к нашему docker-compose.yml:

version: '3.4'x-mssql-base: &mssql-base  image: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04  restart: always  environment:    ACCEPT_EULA: Y    MSSQL_PID: Express    SA_PASSWORD: SayThanksToSatiaAt2020    MSSQL_MEMORY_LIMIT_MB: 1024services:  dwh:    image: jbfavre/vertica:9.2.0-7_ubuntu-16.04  mssql_0:    <<: *mssql-base  mssql_1:    <<: *mssql-base  mssql_2:    <<: *mssql-base  mssql_init:    image: mio101/py3-sql-db-client-base    command: python3 ./mssql_init.py    depends_on:      - mssql_0      - mssql_1      - mssql_2    environment:      SA_PASSWORD: SayThanksToSatiaAt2020    volumes:      - ./mssql_init.py:/mssql_init.py      - ./dags/commons/datasources.py:/commons/datasources.py

Там мы поднимаем:

• Vertica как хост dwh с самыми дефолтными настройками,
• три экземпляра SQL Server,
• наполняем базы в последних кое-какими данными (ни в коем случае не заглядывайте в mssql_init.py!)

Запускаем всё добро с помощью чуть более сложной, чем в прошлый раз, команды:

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

Что нагенерировал наш чудорандомайзер, можно, воспользовавшись пунктом Data Profiling/Ad Hoc Query:

Главное, не показывать это аналитикам

Подробно останавливаться на ETL-сессиях я не буду, там всё тривиально: делаем базу, в ней табличку, оборачиваем всё менеджером контекста, и теперь делаем так:

with Session(task_name) as session:    print('Load', session.id, 'started')    # Load worflow    ...    session.successful = True    session.loaded_rows = 15

from sys import stderrclass Session:    """ETL workflow session    Example:        with Session(task_name) as session:            print(session.id)            session.successful = True            session.loaded_rows = 15            session.comment = 'Well done'    """    def __init__(self, connection, task_name):        self.connection = connection        self.connection.autocommit = True        self._task_name = task_name        self._id = None        self.loaded_rows = None        self.successful = None        self.comment = None    def __enter__(self):        return self.open()    def __exit__(self, exc_type, exc_val, exc_tb):        if any(exc_type, exc_val, exc_tb):            self.successful = False            self.comment = f'{exc_type}: {exc_val}\n{exc_tb}'            print(exc_type, exc_val, exc_tb, file=stderr)        self.close()    def __repr__(self):        return (f'<{self.__class__.__name__} '                 f'id={self.id} '                 f'task_name="{self.task_name}">')    @property    def task_name(self):        return self._task_name    @property    def id(self):        return self._id    def _execute(self, query, *args):        with self.connection.cursor() as cursor:            cursor.execute(query, args)            return cursor.fetchone()[0]    def _create(self):        query = """            CREATE TABLE IF NOT EXISTS sessions (                id          SERIAL       NOT NULL PRIMARY KEY,                task_name   VARCHAR(200) NOT NULL,                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,                finished    TIMESTAMPTZ           DEFAULT current_timestamp,                successful  BOOL,                loaded_rows INT,                comment     VARCHAR(500)            );            """        self._execute(query)    def open(self):        query = """            INSERT INTO sessions (task_name, finished)            VALUES (%s, NULL)            RETURNING id;            """        self._id = self._execute(query, self.task_name)        print(self, 'opened')        return self    def close(self):        if not self._id:            raise SessionClosedError('Session is not open')        query = """            UPDATE sessions            SET                finished    = DEFAULT,                successful  = %s,                loaded_rows = %s,                comment     = %s            WHERE                id = %s            RETURNING id;            """        self._execute(query, self.successful, self.loaded_rows,                      self.comment, self.id)        print(self, 'closed',              ', successful: ', self.successful,              ', Loaded: ', self.loaded_rows,              ', comment:', self.comment)class SessionError(Exception):    passclass SessionClosedError(SessionError):    pass

Настала пора забрать наши данные из наших полутора сотен таблиц. Сделаем это с помощью очень незатейливых строчек:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()query = f"""    SELECT         id, start_time, end_time, type, data    FROM dbo.Orders    WHERE        CONVERT(DATE, start_time) = '{dt}'    """df = pd.read_sql_query(query, source_conn)

1. С помощью хука получим из Airflow pymssql-коннект
2. В запрос подставим ограничение в виде даты в функцию её подбросит шаблонизатор.
3. Скармливаем наш запрос pandas, который достанет для нас DataFrame он нам пригодится в дальнейшем.

Я использую подстановку {dt} вместо параметра запроса %s не потому, что я злобный Буратино, а потому что pandas не может совладать с pymssql и подсовывает последнему params: List, хотя тот очень хочет tuple.
Также обратите внимание, что разработчик pymssql решил больше его не поддерживать, и самое время съехать на pyodbc.

Посмотрим, чем Airflow нашпиговал аргументы наших функций:

Если данных не оказалось, то продолжать смысла нет. Но считать заливку успешной тоже странно. Но это и не ошибка. А-а-а, что делать?! А вот что:

if df.empty:    raise AirflowSkipException('No rows to load')

AirflowSkipException скажет Airflow, что ошибки, собственно нет, а таск мы пропускаем. В интерфейсе будет не зеленый и не красный квадратик, а цвета pink.

Подбросим нашим данным несколько колонок:

df['etl_source'] = src_schemadf['etl_id'] = session.iddf['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

А именно:

• БД, из которой мы забрали заказы,
• Идентификатор нашей заливающей сессии (она будет разной на каждый таск),
• Хэш от источника и идентификатора заказа чтобы в конечной базе (где всё ссыпется в одну таблицу) у нас был уникальный идентификатор заказа.

Остался предпоследний шаг: залить всё в Vertica. А, как ни странно, один из самых эффектных эффективных способов сделать это через CSV!

# Export data to CSV bufferbuffer = StringIO()df.to_csv(buffer,          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,          header=False, float_format='%.8f', doublequote=False, escapechar='\\')buffer.seek(0)# Push CSVtarget_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()copy_stmt = f"""    COPY {target_table}({df.columns.to_list()})     FROM STDIN     DELIMITER '|'     ENCLOSED '"'     ABORT ON ERROR     NULL 'NUL'    """cursor = target_conn.cursor()cursor.copy(copy_stmt, buffer)

1. Мы делаем спецприёмник StringIO.
2. pandas любезно сложит в него наш DataFrame в виде CSV-строк.
3. Откроем соединение к нашей любимой Vertica хуком.
4. А теперь с помощью copy() отправим наши данные прямо в Вертику!

Из драйвера заберем, сколько строчек засыпалось, и скажем менеджеру сессии, что всё ОК:

session.loaded_rows = cursor.rowcountsession.successful = True

Вот и всё.

На проде мы создаем целевую табличку вручную. Здесь же я позволил себе небольшой автомат:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'create_table_query = f"""    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (         id         INT,         start_time TIMESTAMP,         end_time   TIMESTAMP,         type       INT,         data       VARCHAR(32),         etl_source VARCHAR(200),         etl_id     INT,         hash_id    INT PRIMARY KEY     );"""create_table = VerticaOperator(    task_id='create_target',    sql=[create_schema_query,         create_table_query],    vertica_conn_id=target_conn_id,    task_concurrency=1,    dag=dag)

Я с помощью VerticaOperator() создаю схему БД и таблицу (если их еще нет, естественно). Главное, правильно расставить зависимости:

for conn_id, schema in sql_server_ds:    load = PythonOperator(        task_id=schema,        python_callable=workflow,        op_kwargs={            'src_conn_id': conn_id,            'src_schema': schema,            'dt': '{{ ds }}',            'target_conn_id': target_conn_id,            'target_table': f'{target_schema}.{target_table}'},        dag=dag)    create_table >> load

Подводим итоги

Ну вот, сказал мышонок, не правда ли, теперь
Ты убедился, что в лесу я самый страшный зверь?

Джулия Дональдсон, Груффало

Думаю, если бы мы с моими коллегами устроили соревнование: кто быстрее составит и запустит с нуля ETL-процесс: они со своими SSIS и мышкой и я с Airflow А потом бы мы еще сравнили удобство сопровождения Ух, думаю, вы согласитесь, что я обойду их по всем фронтам!

Если же чуть-чуть посерьезнее, то Apache Airflow за счет описания процессов в виде программного кода сделал мою работу гораздо удобнее и приятнее.

Его же неограниченная расширяемость: как в плане плагинов, так и предрасположенность к масштабируемости даёт вам возможность применять Airflow практически в любой области: хоть в полном цикле сбора, подготовки и обработки данных, хоть в запуске ракет (на Марс, конечно же).

Часть заключительная, справочно-информационная

Грабли, которые мы собрали за вас

• start_date. Да, это уже локальный мемасик. Через главный аргумент дага start_date проходят все. Кратко, если указать в start_date текущую дату, а в schedule_interval один день, то DAG запустится завтра не раньше.

  
  

  start_date = datetime(2020, 7, 7, 0, 1, 2)
  

  
  

  И больше никаких проблем.

  
  

  С ним же связана и еще одна ошибка выполнения: Task is missing the start_date parameter, которая чаще всего говорит о том, что вы забыли привязать к оператору даг.

  
  

• Всё на одной машине. Да, и базы (самого Airflow и нашей обмазки), и веб-сервер, и планировщик, и воркеры. И оно даже работало. Но со временем количество задач у сервисов росло, и когда PostgreSQL стал отдавать ответ по индексу за 20 с вместо 5 мс, мы его взяли и унесли.

  
  

• LocalExecutor. Да, мы сидим на нём до сих пор, и мы уже подошли к краю пропасти. LocalExecutorа нам до сих пор хватало, но сейчас пришла пора расшириться минимум одним воркером, и придется поднапрячься, чтобы переехать на CeleryExecutor. А ввиду того, что с ним можно работать и на одной машиной, то ничего не останавливает от использования Celery даже не сервере, который естественно, никогда не пойдет в прод, чесслово!

  
  

• Неиспользование встроенных средств:

  
  
  • Connections для хранения учетных данных сервисов,
  • SLA Misses для реагирования на таски, которые не отработали вовремя,
  • XCom для обмена метаданными (я сказал метаданными!) между тасками дага.
  
  

• Злоупотребление почтой. Ну что тут сказать? Были настроены оповещения на все повторы упавших тасков. Теперь в моём рабочем Gmail >90k писем от Airflow, и веб-морда почты отказывается брать и удалять больше чем по 100 штук за раз.

  
  

Больше подводных камней: Apache Airflow Pitfails

Средства ещё большей автоматизации

Для того чтобы нам еще больше работать головой, а не руками, Airflow заготовила для нас вот что:

• REST API он до сих пор имеет статус Experimental, что не мешает ему работать. С его помощью можно не только получать информацию о дагах и тасках, но остановить/запустить даг, создать DAG Run или пул.

  
  

• CLI через командную строку доступны многие средства, которые не просто неудобны в обращении через WebUI, а вообще отсутствуют. Например:

  
  
  • backfill нужен для повторного запуска инстансов тасков.
    Например, пришли аналитики, говорят: А у вас, товарищ, ерунда в данных с 1 по 13 января! Чини-чини-чини-чини!. А ты такой хоба:
    

    airflow backfill -s '2020-01-01' -e '2020-01-13' orders
    

  • Обслуживание базы: initdb, resetdb, upgradedb, checkdb.
  • run, который позволяет запустить один инстанс таска, да еще и забить на всё зависимости. Более того, можно запустить его через LocalExecutor, даже если у вас Celery-кластер.
  • Примерно то же самое делает test, только еще и в баз ничего не пишет.
  • connections позволяет массово создавать подключения из шелла.
  
  

• Python API довольно хардкорный способ взаимодействия, который предназначен для плагинов, а не копошения в нём ручёнками. Но кто ж нам помешает пойти в /home/airflow/dags, запустить ipython и начать беспредельничать? Можно, например, экспортировать все подключения таком кодом:

  
  

  from airflow import settingsfrom airflow.models import Connectionfields = 'conn_id conn_type host port schema login password extra'.split()session = settings.Session()for conn in session.query(Connection).order_by(Connection.conn_id):  d = {field: getattr(conn, field) for field in fields}  print(conn.conn_id, '=', d)
  

  
  

• Подключение к базе метаданных Airflow. Писать в неё я не рекомендую, а вот доставать состояния тасков для различных специфических метрик можно значительно быстрее и проще, чем через любой из API.

  
  

  Скажем, далеко не все наши таски идемпотентны, а могут иногда падать и это нормально. Но несколько завалов это уже подозрительно, и надо бы проверить.

  
  
  Осторожно, SQL!

  WITH last_executions AS (SELECT    task_id,    dag_id,    execution_date,    state,        row_number()        OVER (            PARTITION BY task_id, dag_id            ORDER BY execution_date DESC) AS rnFROM public.task_instanceWHERE    execution_date > now() - INTERVAL '2' DAY),failed AS (    SELECT        task_id,        dag_id,        execution_date,        state,        CASE WHEN rn = row_number() OVER (            PARTITION BY task_id, dag_id            ORDER BY execution_date DESC)                 THEN TRUE END AS last_fail_seq    FROM last_executions    WHERE        state IN ('failed', 'up_for_retry'))SELECT    task_id,    dag_id,    count(last_fail_seq)                       AS unsuccessful,    count(CASE WHEN last_fail_seq        AND state = 'failed' THEN 1 END)       AS failed,    count(CASE WHEN last_fail_seq        AND state = 'up_for_retry' THEN 1 END) AS up_for_retryFROM failedGROUP BY    task_id,    dag_idHAVING    count(last_fail_seq) > 0
  

  
  
  
  

Ссылки

Ну и естественно первые десять ссылок из выдачи гугла содержимое папки Airflow из моих закладок.

• Apache Airflow Documentation конечно, надо начать с оф. документации, но кто же читает инструкции?
• Best Practices ну хотя бы рекомендации от создателей прочитайте.
• The Airflow UI самое начало: пользовательский интерфейс в картинках
• Understanding Apache Airflows key concepts хорошо расписаны базовые понятия, если (вдруг!) вы что-то не поняли у меня.
• Tianlong's Blog A Guide On How To Build An Airflow Server/Cluster краткий гайд по настройке Airflow-кластера.
• Running Apache Airflow At Lyft почти такая же интересная статья, разве что формализма побольше, а примеров поменьше.
• How Apache Airflow Distributes Jobs on Celery workers о работе в связке с Celery.
• DAG Writing Best Practices in Apache Airflow про идемпотентность тасков, загрузку по ID вместо даты, трансформации, структуру файлов и прочие интересные вещи.
• Managing Dependencies in Apache Airflow зависимости тасков и Trigger Rule, которые я упомянул лишь вскользь.
• Airflow: When Your DAG is Far Behind The Schedule как преодолевать некоторые работает, как задумано у планировщика, загружать потерянные данные и расставлять приоритеты тасков.
• Useful SQL queries for Apache Airflow полезные SQL-запросы к метаданным Airflow.
• Get started developing workflows with Apache Airflow есть полезный раздел про создание кастомного сенсора.
• Building the Fetchr Data Science Infra on AWS with Presto and Airflow интересная короткая заметка о построении инфраструктуры на AWS для Data Science.
• 7 Common Errors to Check when Debugging Airflow DAGs распространенные ошибки (когда кое-кто всё-таки не читает инструкций).
• Store and access password using Apache Airflow улыбнитесь, как люди костылят хранение паролей, хотя можно просто использовать Connections.
• The Zen of Python and Apache Airflow неявный проброс DAG, заброс контекста в функции, снова про зависимости, а еще про пропуск запусков тасков.
• Airflow: Lesser Known Tips, Tricks, and Best Practises об использовании default arguments и params в шаблонах, а также о Variables и Connections.
• Profiling the Airflow Scheduler рассказ о том, как планировщик готовят к Airflow 2.0.
• Apache Airflow with 3 Celery workers in docker-compose немножко устаревшая статья про деплой нашего кластера в docker-compose.
• 4 Templating Tasks Using the Airflow Context динамические таск с помощью шаблонов и проброса контекста.
• Error Notifications in Airflow стандартные и кастомные оповещения почтой и Slack.
• Airflow Workshop: сложные DAGи без костылей Ветвления тасков, макросы и XCom.

И ссылки, задействованные в статье:

• Macros reference доступные для использования в шаблонах плейсхолдеры.
• Common Pitfalls Airflow Распространенные ошибки при создании дагов.
• puckel/docker-airflow: Docker Apache Airflow docker-compose для экспериментов, отладки и не только.
• python-telegram-bot/python-telegram-bot: We have made you a wrapper you can't refuse Python-обертка для Telegram REST API.

Проработав фулстек разработчиком около 10 лет, я заметил одну странность.
Я ни разу не встретил не MVC веб-фреймворк. Да, периодически встречались вариации, однако общая структура всегда сохранялась:

• Codeigniter мой первый фреймворк, MVC
• Kohana MVC
• Laravel MVC
• Django создатели слегка подменили термины, назвав контроллер View, а View Template'ом, но суть не изменилась
• Flask микрофреймворк, по итогу все равно приходящий к MVC паттерну
  

Конечно, с моим мнением можно поспорить, можно продолжить перечислять, однако суть не в этом.

Меня беспокоило то, что за все время существования веб-разработки, MVC является, по сути, монополистом в проектировании приложений. Я не говорю что это плохо,
просто это казалось мне странным.

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

1. REST (порой GraphQL или другие варианты) бэкенд, выполняющий роль провайдера данных.
2. Frontend, написаный на каком-либо из фреймворков большой тройки.

Задачи, которые сейчас стоят перед бэкендом (если сильно упростить) это взять данные из базы, преобразовать в JSON (возможно дополнительно преобразовав структуру) и отправить в браузер.

В ходе этих размышлений, мой взгляд упал на ETL паттерн, и в определенный момент я понял, что он идеально подходит для всех задач, которые на данный момент стоят перед бэкендом.
Осознав это, я решил провести эксперимент, и результатом этого эксперимента стал Pipe Framework.

О фреймворке

В Pipe Framework (далее PF) нет понятий модель-представление-контроллер, но я буду использовать их для демонстрации его принципов.

Весь функционал PF строится с помощью "шагов" (далее Step).

Step это самодостаточная и изолированная единица, призванная выполнять только одну функцию, подчиняясь принципу единственной ответственности (single responsibility principle).

Более детально объясню на примере. Представим, у вас есть простая задача создать API ендпоинт для todo приложения.

При традиционном подходе, вам необходимо создать Todo модель, которая представляет собой таблицу в базе данных.
В контроллере, привязанном к роуту, вы будете использовать экземпляр модели, чтобы извлечь данные о todo тасках, трансформировать их в https ответ, и отправить пользователю.

Я выделил извлечь и трансформировать чтобы вы могли ассоциировать MVC концепты с концептами, которые я использую в PF.

То есть, мы можем провести аналогию между MVC (Модель-Представление-Контроллер) и ETL (Извлечение-Преобразование-Загрузка):

Model Extractor / Loader

Controller Transformer

View Loader

Эта довольно приблизительная аналогия, однако она показывает как части одного и другого подхода связаны друг с другом.

Как видите, я обозначил View как Loader. Позже станет понятно, почему я так поступил.

Первый роут

Давайте выполним поставленную задачу используя PF.

Первое, на что необходимо обратить внимание, это три типа шагов:

• Extractor
• Transformer
• Loader

Как определиться с тем, какой тип использовать?

1. Если вам надо извлечь данные из внешнего ресурса: extractor.
2. Если вам надо передать данные за пределы фреймворка: loader.
3. Если вам надо внести изменения в данные: transformer.

Именно поэтому я ассоциирую View с Loader'ом в примере выше. Вы можете воспринимать это как загрузку данных в браузер пользователя.

Любой шаг должен наследоваться от класса Step, но в зависимости от назначения реализовывать разные методы:

class ESomething(Step):    def extract(self, store):        ...class TSomething(Step):    def transform(self, store):        ...class LSomething(Step):    def load(self, store):        ...

Как вы можете заметить, названия шагов начинаются с заглавных E, T, L.
В PF вы работаете с экстракторами, трансформерами, и лоадерами, названия которых слишком длинные, если использовать их как в примере:

class ExtractTodoFromDatabase(Extractor):    pass

Именно поэтому, я сокращаю названия типа операции до первой буквы:

class ETodoFromDatabase(Extractor):    pass

E значит экстрактор, T трансформер, и L лоадер.
Однако, это просто договоренность и никаких ограничений со стороны фреймворка нет, так что можете использовать те имена, которые захотите :)

Для того что бы выполнить задачу, прежде всего нам нужно декомпозировать функционал на более мелкие операции:

1. Извлекаем данные из базы
2. Преобразовываем данные в JSON
3. Отправляем данные в браузер посредством HTTP.

Итак, нам нужен будет 1 экстратор, 1 трансформер, и 1 лоадер.
К счастью, в PF есть набор предопределенных шагов, и они полностью покрывают описаные выше операции. Но, тем не менее, нам все-таки придется создать экстрактор, потому что нужно будет прописать данные доступа к базе данных.

Так как шаг является независимой частью приложения, которая отвечает лишь за одну задачу и не осведомлена обо всей остальной системе, его легко переносить из одного пайпа в другой, из приложения в приложение и т. д.

Недостаток такого решения: отсутствие центрального хранилища конфигурации. Все конфиги, относящиеся к определенному шагу, должны храниться в свойствах класса шага. Порой, это значит то, что нам необходимо писать один и тот же конфиг каждый раз при работе с шагами с одинаковой конфигурацией.

Для этих целей, в PF предусмотрен @configure декоратор. То есть, вы просто перечисляете настройки, которые хотите добавить в шаг, следующим образом:

DATABASES = {    'default': {        'driver': 'postgres',        'host': 'localhost',        'database': 'todolist',        'user': 'user',        'password': '',        'prefix': ''    }}DB_STEP_CONFIG = {    'connection_config': DATABASES}

и потом передаете как аргумент декоратору, примененному к классу:

@configure(DB_STEP_CONFIG)class EDatabase(EDBReadBase):    pass

Итак, давайте создадим корневую папку проекта:

pipe-sample/

Затем папку src внутри pipe-sample:

pipe-sample/    src/

Все шаги, связанные с базой данных, будут находится в db пакете, давайте создадим и его тоже:

pipe-sample/    src/        db/            __init__.py

Создайте config.py файл с настройками для базы данных:

pipe-sample/src/db/config.py

DATABASES = {    'default': {        'driver': 'postgres',        'host': 'localhost',        'database': 'todolist',        'user': 'user',        'password': '',        'prefix': ''    }}DB_STEP_CONFIG = {    'connection_config': DATABASES}

Затем, extract.py файл для сохранения нашего экстрактора и его концигурации:

pipe-sample/src/db/extract.py

from src.db.config import DB_STEP_CONFIG # наша конфигурация"""PF включает в себя несколько дженериков для базы данных,которые вы можете посмотреть в API документации"""from pipe.generics.db.orator_orm.extract import EDBReadBase@configure(DB_STEP_CONFIG) # применяем конфигурацию к шагу class EDatabase(EDBReadBase):    pass     # нам не надо ничего добавлять внутри класса    # вся логика уже имплементирована внутри EDBReadBase

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

Теперь мы готовы к созданию первого пайпа.

Добавьте app.py в корневую папку проекта. Затем скопируйте туда этот код:

pipe-sample/app.py

from pipe.server import HTTPPipe, appfrom src.db.extract import EDatabasefrom pipe.server.http.load import LJsonResponse from pipe.server.http.transform import TJsonResponseReady@app.route('/todo/') # декоратор сообщает WSGI приложению, что этот пайп обслуживает данный маршрутclass TodoResource(HTTPPipe):     """    мы расширяем HTTPPipe класс, который предоставляет возможность описывать схему пайпа с учетом типа HTTP запроса    """    """    pipe_schema это словарь с саб пайпами для каждого HTTP метода.     'in' и 'out' это направление внутри пайпа, когда пайп обрабатывает запрос,    он сначала проходит через 'in' и затем через 'out' пайпа.    В этом случае, нам ничего не надо обрабатывать перед получением ответа,     поэтому опишем только 'out'.    """    pipe_schema = {         'GET': {            'out': (                # в фреймворке нет каких либо ограничений на порядок шагов                # это может быть ETL, TEL, LLTEETL, как того требует задача                # в этом примере просто так совпало                EDatabase(table_name='todo-items'),                TJsonResponseReady(data_field='todo-items_list'), # при извлечении данных EDatabase всегда кладет результат запроса в поле {TABLE}_item для одного результата и {TABLE}_list для нескольких                LJsonResponse()            )        }    }"""Пайп фреймворк использует Werkzeug в качестве WSGI-сервера, так что аргументы должны быть знакомы тем кто работал, например, с Flask. Выделяется только 'use_inspection'. Inspection - это режим дебаггинга вашего пайпа.Если установить параметр в True до начала воспроизведения шага, фреймворк будет выводить название текущего шага и содержимое стор на этом этапе."""if __name__ == '__main__':    app.run(host='127.0.0.1', port=8080,            use_debugger=True,            use_reloader=True,            use_inspection=True            )

Теперь можно выполнить $ python app.py и перейти на http://localhost:8000/todo/.

Из примера выше довольно сложно понять как выглядит реализация шага, поэтому ниже я приведу пример из исходников:

class EQueryStringData(Step):    """    Generic extractor for data from query string which you can find after ? sign in URL    """    required_fields = {'+{request_field}': valideer.Type(PipeRequest)}    request_field = 'request'    def extract(self, store: frozendict):        request = store.get(self.request_field)        store = store.copy(**request.args)        return store

Стор

На данный момент, стор в PF это инстанс frozendict.
Изменить его нельзя, но можно создать новый инстанс используя frozendict().copy() метод.

Валидация

Мы помним, что шаги являются самостоятельными единицами функционала, но иногда они могут требовать наличия определенных данных в сторе для выполнения каких-либо операций (например id пользователя из URL). В этом случае, используйте поле required_fields в конфигурации шага.

PF использует Valideer для валидации. На данный момент, я рассматриваю альтернативы, однако в случае смены библиотеки принцип останется тот же.

Пример

Все, что нам надо сделать это написать dict с необходимыми полями в теле шага (здесь вы найдете больше информации о доступных валидаторах: Valideer).

class PrettyImportantTransformer(Step):    required_fields = {'+some_field': valideer.Type(dict)} # `+` значит обязательное поле

Динамическая валидация

Иногда, в шаге у вас может быть переменная, которая хранит название ключа в сторе, по которому можно найти необходимую информацию.
Вы не можете узнать, как именно называется это поле, но знаете как называется переменная в шаге, которая хранит эти данные.
Если вы хотите валидировать и эти поля, необходимо добавить фигурные скобки с названием переменной класса:

class EUser(Step):    pk_field = 'id' # EUser будет обращаться к полю 'id' в сторе    required_fields = {'+{pk_field}': valideer.Type(dict)} # все остальное так же

Пайп фреймворк заменит это поле на значение pk_field автоматически, и затем валидирует его.

Объединение шагов

Вы можете объединить два или более шага в случае, если вам необходимо контролировать порядок выполнения.

В этом примере я использую оператор | (OR)

    pipe_schema = {        'GET': {            'out': (                # В случае если EDatabase бросает любое исключение                 # выполнится LNotFound, которому в сторе передастся информация об исключении                EDatabase(table_name='todo-items') | LNotFound(),                 TJsonResponseReady(data_field='todo-items_item'),                LJsonResponse()            )        },

Так же есть оператор & (AND)

    pipe_schema = {        'GET': {            'out': (                # В этом случае оба шага должны выполниться успешно, иначе стор без изменений перейдет к следующему шагу                 EDatabase(table_name='todo-items') & SomethingImportantAsWell(),                 TJsonResponseReady(data_field='todo-items_item'),                LJsonResponse()            )        },

Хуки

Чтобы выполнить какие-либо операции до начала выполнения пайпа, можно переопределить метод: before_pipe

class PipeIsAFunnyWord(HTTPPipe):    def before_pipe(self, store): # в аргументы передается initial store. В случае HTTPPipe там будет только объект PipeRequest        pass

Также есть хук after_pipe и я думаю нет смысла объяснять, для чего он нужен.

interrupt это последний из доступных хуков, должен возвращать bool. Вызывается после каждого шага, в качестве аргумента получая текущий стор. В случае, если метод возвращает True, выполнение пайпа заканчивается и он возвращает стор в текущем его состоянии.

Пример использования из исходников фреймворка:

class HTTPPipe(BasePipe):    """Pipe structure for the `server` package."""    def interrupt(self, store) -> bool:        # If some step returned response, we should interrupt `pipe` execution        return issubclass(store.__class__, PipeResponse) or isinstance(store, PipeResponse)

Потенциальные преимущества

Разрабатывая Pipe Framework, я ничего от него не ожидал, однако в ходе работы я смог выделить довольно большое количество преимуществ такого подхода:

1. Принудительная декомпозиция: разработчик вынужден разделять задачу на атомарные шаги. Это приводит к тому, что сначала надо подумать, а потом делать, что всегда лучше, чем наоборот.
2. Абстрактность: фреймворк подразумевает написание шагов, которые можно применить в нескольких местах, что позволяет уменьшить количество кода.
3. Прозрачность: любая, пусть даже и сложная логика, спрятанная в шагах, призвана выполнять понятные для любого человека задачи. Таким образом, гораздо проще объяснить даже нетехническому персоналу о том, что происходит внутри через преобразование данных.
4. Самотестируемость: даже без написаных юнит тестов, фреймворк подскажет вам что именно и в каком месте сломалось за счет валидации шагов.
5. Юнит-тестирование осуществляется гораздо проще, нужно только задать начальные данные для шага или пайпа и проверить, что получается на выходе.
6. Разработка в команде тоже становится более гибкой. Декомпозировав задачу, можно легко распределить различные шаги между разработчиками, что практически невозможно сделать при традиционном подходе.
7. Постановка задачи сводится к предоставлению начального набора данных и демонстрации необходимого набора данных на выходе.
   

Фреймворк на данный момент находится в альфа-тестировании, и я рекомендую экспериментировать с ним, предварительно склонировав с Github репозитория. Установка через pip так же доступна

pip install pipe-framework


Планы по развитию:

1. Django Pipe: специальный тип Pipe, который можно использовать как Django View.
2. Смена Orator ORM на SQL Alchemy для Database Generics (Orator ORM библиотека с приятным синтаксисом, но слабой поддержкой, парой багов, и недостаточным функционалом в стабильной версии).
3. Асинхронность.
4. Улучшеный Inspection Mode.
5. Pipe Builder специальный веб-дашбоард, в котором можно составлять пайпы посредством визуальных инструментов.
6. Функциональные шаги на данный момент шаги можно писать только в ООП стиле, в дальнейшем планируется добавить возможность использовать обычные функции

В целом, планируется двигать фреймворк в сторону упрощения, без потери функциональности. Буду рад вопросам и контрибьюшнам.

Хорошего дня!

В конце зимы и начале весны, появилась возможность поработать с новым для меня инструментом потоковой доставки данных Apache NiFi. При изучении инструмента, все время не покидало ощущение, что помимо официальной документации, нелишним были бы материалы "for dummies", с практическими примерами.

После выполнении задачи, решил попробовать облегчить вхождение в мир NiFi.

Предыстория, почти не связанная со статьей

В феврале этого года поступило предложение разработать прототип системы для обработки данных из разных источников. Источники предоставляют информацию не одновременно, в разных форматах. Нужно было ее преобразовывать в единый формат, собирать из источников информацию в одну структуру и передавать на обработку. При обработке использовались дополнительные справочники из внутренних баз данных для обогащения. После обработки данные должны были записываться в две базы данных и генерироваться суточный отчет.
Задача укладывалась в концепцию ETL. Поиск реализаций ETL привел на довольно большое количество ETL-систем, которые позволяют выстроить процессы обработки. В том числе opensource системы.

Система Apache NiFi была выбрана на удачу. Прототип был построен и сдан заказчику.

Первоначально заказчик хотел монолитное приложение, а использование NiFi рассматривал просто как инструмент прототипирование (где-то прочитал). Но после знакомства в вблизи NiFi остался в продукте.

А теперь собственно история

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

Когда я начал разбираться с Apache NiFi выяснилось, что более-менее подробная информация есть только на сайте проекта. Русские статьи во многом просто переводы вводных частей официальной документации. Основной недостаток часто не понятно в каком формате параметры вводить. Гугл спасает, но не всегда.

И появилась идея написать статью с описанием практического примера работы системы введение для новичков. В комментариях, дорогой читатель, можно высказаться получилось или нет.

И так, задача получать данные о распространении вируса, обрабатывать строить графики.
Источниками данных будет сайты стопкороновирус.рф и covid19.who.int. Первый сайт содержит данные по регионам России, сайт ВОЗ данные по странам мира.

Данные на сайте стопкороновирус.рф находятся прямо в html-коде страницы в виде почти готового json-массива. Нужно его только обработать. Но об этом в следующей статье.
Данные ВОЗ находятся в csv-файле, который не получилось автоматически скачивать (либо я был недостаточно настойчив). Поэтому, когда нужно было, файл сохранялся из браузера в специальную папку.

Если кратко сказать о NiFi система при выполнении процесса (pipeline), состоящего из процессоров, получает данные, модифицирует и куда-то сохраняет. Процессоры выполняют работу читают файлы, преобразовывают один формат в другой, обогащают данные из другой базы данных, загружают в хранилища и т.д. Процессоры между собой соединены очередями. У каждой очереди есть признак, по которому данные в нее загружаются из процессора. Например в одну очередь можно загрузить данные при удачной выполнении работы процессора, в другую при сбое. Это позволяет запустить данные по разным веткам процессов. Например, берем данные, ищем подстроку если нашли отправляем по ветке 1, у тех данных, в которых подстрока не найдена отправляем по ветке 2.

Каждый процессор может находиться в состоянии "Running" или "Stopped". Данные накапливаются в очереди перед остановленным процессором, после старта процессора, данные будут переданы в процессор. Очень удобно, можно изменять параметры процессора, без потери данных на работающем процессе.

Данные в системе называются FlowFile, потоковый файл. Процесс начинает работать с процессора, который умеет генерировать FlowFile. Не каждый процессор умеет генерировать потоковый файл в этом случае в начало ставится процессор "GenerateFlowFile", задача которого создать пустой потоковый файл и тем самым запустить процесс. Файл может создаваться по расписанию или событию. Файл может быть бинарным или текстовым потоком.

Файл состоит из контента и метаданных. Метаданные называются атрибутами. Например, атрибутом является имя файл, id-файла, имя схемы и т.д. Система позволяет добавлять собственные атрибуты, записывать значения. Значения атрибутов используются в работе процессоров.

Содержимое файла может быть бинарным или текстовым. Поток данных можно представить в виде записей. Для записей можно определить их структуру схему данных. Схема описывает понимается формат записи, в которой описываются имена и типы полей, валидные значения, значения по умолчанию и т.д.

В NiFi еще одна важная сущность контроллеры. Это объекты которые знают как сделать какую-то работу. Например, процессор хочет записать в базу данных, то контроллер будет знать, как найти и подключиться к базе данных, таблице.

Контролер, который знает как читать данные на входе в контроллер называется Reader. После обработки, на выходе процессора, данные форматируются согласно контроллеру RecordSetWriter.

Итак практическое применение вольного изложения документации чтение данных ВОЗ и сохранение в БД. Данные будем хранить в СУБД PostgreSQL.

Создаем таблицу:

CREATE TABLE public.who_outbreak (    dt timestamp NULL,    country_code varchar NULL,    country varchar NULL,    region_code varchar NULL,    died int4 NULL,    died_delta int4 NULL,    infected int4 NULL,    infected_delta int4 NULL);

В таблице будет храниться дата сбора данных, код и наименование страны, код региона, суммарное количество погибших, количество погибших за сутки, суммарное количество заболевших, количество заболевших за сутки.

Источником данных будет cvs-файл с сайта https://covid19.who.int/ по кнопке Download Map Data. Файл содержит информацию по заболевшим и погибшим по всем странам на каждый день примерно с конца января. Оперативная информация там задерживается на 1-2 дня. За это время в файле менялись наименования полей (были даже наименования с пробелами), менялся формат даты.

Файл сохраняется из браузера в определенный каталог, откуда NiFi забирает его на обработку.

Общий вид визуализации процесса в интерфейсе Apache NiFI

На рисунке большие прямоугольники процессы, прямоугольники поменьше очереди между процессами.

Обработка начинается с верхнего процесса, отходящие от процесса стрелочки показывают направление движения FlowFile (атрибутов и контента).

На визуализации процесса показывается его статус (работает или нет), данное имя процессу, его тип, количество и общий объем прошедших файлов на входе и на выходе за период времени, в данном случае за 5 мин.

Визуализация очереди показывает ее имя, и объем данных в очереди. По умолчанию имя очереди это тип связи success, failed и другие.

Тип первого процесса "GetFile". Этот процессор создает flowfile и запускает процесс. Контентом потокового файла будет содержимое файла если он будет найден. В настройках процессора на вкладке Scheduling указываем расписание запуска процесса 20 секунд.

Вкладка Scheduling запуск каждые 20 сек.

После старта процессора, каждые 20 секунд процессор будет запускаться. Если файл будет найден FlowFile будет создан и процесс запустится.

Как видно из рисунка, указываем каталог и имя файла. В имени файла можно использовать символы подстановки. Например *.csv приведет к обработке всех csv-файлов в каталоге. Указываем также, что после обработки файл можно удалить ("Keep Source File"). Также есть возможность указать максимальные и минимальные значения возраста и размера файла. Это позволяет обрабатывать, например, только не пустые файл, созданные за последний час.
На вкладке Settings указываются базовые параметры процесса, такие как имя процесса, максимальное время работы процесса, время между запусками, типы связей.

Результатом работы первого процесса "GetFile" с именем "Read WHO datafile" будет просто поток данных из файла. Поток будет передан в следующий процесс "ReplaceText".

Процессор поиска подстроки

В этом процессе обратим внимание сразу на вкладку параметров. Данный процессор ищет regex-выражение "Search Value" в входном потоке и заменяет на новое значение "Replacement Value". Обработка ведется построчно ("Evaluation Mode"). В данном случае идет замена в строке даты. Во входном файле, в какой-то момент дата формата YYYY-MM-DD стала указываться как YYYY-MM-DDThh:mm:ssZ, причем время было всегда 00:00:00, а временная зона не указывалась.
Простого способа преобразования в даты уже в записи не нашел, поэтому к проблеме подошел в лоб просто через процессор "ReplaceText" убрал символы T и Z. После этого строка стала конвертироваться в timestamp в avro-схеме без ошибок.

На выходе процессора будет поток текстовых данных, в которых уже поправили подстроку даты. Но пока это просто поток байтов без какой-то структуры.

Следующий процессор "Rename fields" читает поток уже как структурированные данные.

Переименование полей

Процессор содержит ссылку на Reader специальный объект-контроллер, который умеет читать из потока структурированные данные и в таком виде уже передает процессору на обработку. В данном случае "WHO CVS Reader просто читает поток и преобразует каждую строку cvs-файла в запись (record) которая содержит поля со значениями из строки. Имена полей берутся из заголовка cvs-файла.

Контроллер чтения записей из cvs-файла

Параметр "Schema Access Strategy" указывают, что структура записи формируется из заголовка cvs-файла. Если заголовков нет, то можно изменить стратегию доступа к схеме и в реестре схем данных создать схему, указать ее имя в параметре "Schema Name" или еще проще указать саму схему в параметре "Schema Text".

Но так как у нас есть заголовки в файле читаем по ним.

Итак, в процессоре данные уже структурированные их можно представить как таблицу базы данных поля и строки. И к этой таблице мы выполняем SQL запрос :

select    Date_reported dt,    Country_code country_code,    Country country,    WHO_region region_code,    New_deaths died_delta,    Cumulative_deaths died,    New_cases infected_delta,    Cumulative_cases infectedfrom FLOWFILE

Поля в запросе такие же как заголовки cvs-файла. Имя таблицы служебное FLOWFILE обозначает чтение структурированных данных их контента файла. Язык запроса SQL довольно гибкий, есть функции преобразований, агрегаций и т.д. В данном случае запрос выводит все данные, только имена полей результата будут другие они соответствуют полям таблицы who_outbreak в целевой БД.

Поток записей с новыми именами полей передается в контроллер RecordSetWriter, ссылка на который также указана в параметра контроллера "WHO AvroRecordSetWriter".

Контроллер RecordSetWriter

Контролер RecordSetWriter уже использует предопределенную схему данных. Схема находится в отдельном объекте регистре схем ("Schema Registry"). В контроллере есть только ссылка на реестр схем и имя схемы.

Регистр схем

Работать с регистром схем довольно просто. Добавляем новый параметр. Его имя будет именем схемы. Значение параметр определение схемы.

В регистре схем создана схема who_outbreak, определение схемы:

{"type" : "record","name" : "who_outbreak","fields": [  {"name": "dt", "type": { "type" : "long", "logicalType": "timestamp-millis"}},  {"name": "country_code", "type" : ["null", "string"], "default": "-"},  {"name": "country", "type" : ["null", "string"], "default": ""},  {"name": "region_code", "type" : ["null", "string"], "default": ""},  {"name" : "died", "type" : "int", "default": 0},  {"name" : "died_delta", "type" : "int", "default": 0},  {"name" : "infected", "type" : "int", "default": 0},  {"name" : "infected_delta", "type" : "int", "default": 0} ]}

Имена и типы атрибутов схемы соответствуют именам и типам полей записи, сформированной sql-запросом.

После выполнения контроллером sql-запроса и передачи данных на выход контроллера в формате схемы данных, структурированный поток передается в контроллер "Delete all records". Это контроллер типа "PutSQL", который может передавать на выполнение sql-команды.

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

delete from who_outbreak;

В параметрах контроллера указываем SQL Statement delete from who_outbreak; и ссылку на пул соединений "JDBC Connection Pool". Параметры JDBC стандартные. Пул содержит настройки подключения к конкретной БД, поэтому его можно использовать во всех контроллерах, которые будут работать с этой БД.

Данные или атрибуты FlowFile не обрабатываются в процессоре, поэтому вход и выход процессора идентичен.

Последний процессор "PutDatabaseRecord".

Запись в БД

В этом процессоре указываем Reader, в котором используется определенная схема who_outbreak. Так как мы удалили все записи в предыдущем процессоре, используем простой INSERT для добавления записей в таблицу. Указываем пул соединений DBCPConnectionPool, далее указываем БД и имя таблицы. Имена полей в схеме данных и БД совпадают, то больше никакой дополнительной настройки проводить не нужно.

Все процессоры, контроллеры и регистры схем нужно перевести в состояние Running (Start).
Процесс доставки данных готов. Если положить файл WHO-COVID-19-global-data.csv в каталог D:\input, то в течении 20 секунд он будет удален, а данные пройдя через процесс доставки данных будут сохранены в БД.

Сейчас вынашиваю планы на вторую часть статьи, в которой планирую описать второй процесс ETL чтение данных с сайта, обогащение, загрузка в БД, в файлы, расщепление потока на несколько потоков и сохранение в файлы. Если тема интересна пишите в комментарии, это мне прибавит стимула быстрее написать.

На рисунке изображение в интерфейсе Apache NiFi описанного процесса (справа) и, для затравки, процесса для второй статьи (слева).