From 919dee8e8bb01d401afef38825258543a276fa31 Mon Sep 17 00:00:00 2001
From: Dmitry Dementev <me@dementev.space>
Date: Sun, 12 Oct 2025 17:46:13 +0300
Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=BE=20=D0=B8=D1=82=D0=BE=D0=B3=D0=B0?=
 =?UTF-8?q?=D0=BC=20=D1=82=D0=B5=D1=81=D1=82=D0=B8=D1=80=D0=BE=D0=B2=D0=B0?=
 =?UTF-8?q?=D0=BD=D0=B8=D1=8F?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 airflow-greenplum/.env.example                 |   3 ++-
 airflow-greenplum/README.md                    |   2 ++
 .../__pycache__/greenplum.cpython-312.pyc      | Bin 0 -> 5426 bytes
 .../airflow/dags/kafka_to_greenplum.py         |  17 ++++++++---------
 4 files changed, 12 insertions(+), 10 deletions(-)
 create mode 100644 airflow-greenplum/airflow/dags/helpers/__pycache__/greenplum.cpython-312.pyc

diff --git a/airflow-greenplum/.env.example b/airflow-greenplum/.env.example
index fb6257a..7d5f63a 100644
--- a/airflow-greenplum/.env.example
+++ b/airflow-greenplum/.env.example
@@ -17,7 +17,8 @@ GP_CONN_ID=greenplum_conn
 GP_USE_AIRFLOW_CONN=true
 
 # Kafka Configuration
-KAFKA_BOOTSTRAP=kafka:9092
+KAFKA_BOOTSTRAP=kafka:29092
 KAFKA_TOPIC=orders
 KAFKA_BATCH_SIZE=500
 KAFKA_POLL_TIMEOUT=10
+KAFKA_MAX_EMPTY_POLLS=3
diff --git a/airflow-greenplum/README.md b/airflow-greenplum/README.md
index 8311054..e9a4280 100644
--- a/airflow-greenplum/README.md
+++ b/airflow-greenplum/README.md
@@ -77,10 +77,12 @@ airflow connections add 'greenplum_conn' \
 - Greenplum: `localhost:${GP_PORT:-5432}` (внешний порт проброшен из контейнера)
 - Postgres (Airflow metadata): `localhost:5433`
 - Kafka (для клиентов на хосте): `localhost:9092`
+- Kafka (из контейнеров Docker): `kafka:29092`
 
 ### Параметры чтения/загрузки
 - `KAFKA_BATCH_SIZE` — размер батча при вставке в Greenplum (по умолчанию 500).
 - `KAFKA_POLL_TIMEOUT` — таймаут ожидания сообщения в секундах (по умолчанию 10).
+- `KAFKA_MAX_EMPTY_POLLS` — сколько подряд пустых `poll` допускается перед выходом из цикла (по умолчанию 3).
 
 ### Проверка загрузки
 ```bash
diff --git a/airflow-greenplum/airflow/dags/helpers/__pycache__/greenplum.cpython-312.pyc b/airflow-greenplum/airflow/dags/helpers/__pycache__/greenplum.cpython-312.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..5098ff6518cc744d8f9867562a15364c20c0cb01
GIT binary patch
literal 5426
zcmdT|TWlQF89p<6_wIVVwqs)(!XyOng0oQqp<I%H6K`%x)~=lkF(RwY&e&dOcV~NM
z#`apKG7gtmK{cVMiB!=>JOzb^kR}i)7|Ih>RcQuCA|1IGO8Zdt#f>AS@>Bo+%+6k7
zL)sUl&T9X2&Yb_;zwiJ5GrtQ4YY047uKsgK4iWMV7VO7qWw^x~kU2vR5|M~phGe;Z
zj>9~k;rn?=9T`W~+3(D{`dwLfzq?rH>G$wN;)m<(?#ij?6kQj1tK9GVRxP*av9*R*
zm0R1n3&o7+9U=YxK_dFz=K2F-jT999TcA!1yhQqI#YQpslCwV~t`ch@t%Gt1(y-Vh
z)<Iei<uIfTP_Box5y}mat}5`6=6~P_5zh1{b2(W}s%bf=#3K&Vxi77#rngV}ktXF*
zlIc32<uj65v#TeuqdOi??CCOV2dO0G@)<3gNXfaJ*#LD1`(laCJ-s{kbsuJWrc<Sw
z1pRRvDV~DqC`d}g8R1Q@Oht(%(xT};mL5#!)bcx38E;e5SxHfo*}O2CR)=Iw74VHP
zF6Sgr^rtjMm9wlpt;H4araPIHwVY~tavD^mQ{$^}3D2>4Kc}d)Ljdvo^JY`**`C;r
z1F^0|cW+m$w=dDRV|Q$SX9_pMegcqj5W+1^aR>|5eoyE+9{>bk!Daau2uFA_K~%S0
z{jTEz=;QMOH^B}2tbC=k544Y{LHqqMF5tFD4cjk<>+M`QtpwK{G~Hzw-a*qNdTl7)
zLDMJJlr&4Ay+a$n7_i&ed-QJbg3DS{AGwdHjHFmzN2r4%N62e$jFK_O5i-g}YPC_Q
z(BIH!^$+w*#-x7HnAR`qSBxnkuM|>pesGh37?e^7U_t*#*kxl=Xg6L3#Yz2&{(;b$
zrUMyyRM=6WdsY9nepS#vf&M-<CXE^WW8;MJ6G8t}pA};9Ly;&&8UsZFJ+h(#=9S&D
zd>qPk`C^&DACc?=26SyVJ%jnAm`&$QkHv^y(+`=R&c43G06CMe_q)6M4w$|Yh1k2E
z?%o5YbMvE*Y>K!{w|Fd<%u1$HQzUA7^GQV+1<shxAsE(l=4Glz{1h=}Mv~ULqIoKh
zq=8mS)Sg>(2qs%DQamr}!A9OeNwtXUi;bnEJnkF}Q`}C|Q`}FJmle|ux>9b$bTFQ>
zPH-O3GwzKbEIqJ6&Z`@W)7T&;2bB#&QYH^G*-%<yG+!|NIOt%WjY~H`Tbz$_2E-)!
zyQg)5_xQqp48*@~xc74FwT3MxyRJ7hpV@kP>rbE718Z;AkU-;C{x-wkHXjPl6s8Mj
zo6kk&LLKwL(9DtPBd-^r6zrG}H=JoX-Ewy8`L%Q5_2s&U=YrAsVBMSbtoDg>15gXE
zZu;E;_X2Ud-2cAmBXun|2=5Dj9jIGyLAs<M5l^n&zKMLk$+vx*<MXW?$W)Q$P)7*T
zpBanukU0YeVYspuHQ=#{zu7Vkpnv-#ayBtI0>-0AjY=)Mk4e9ZC&(VLWOAlBFikxE
zY@%_bz~4tU5``P(kx(4`3Ve}LPHjJQp#Kt>F{@ufh5+=DP>cyc382kY{fb~r0Wsb~
zju@{PrvzrT6+yow7%v(x1F1d+oxFA|lTJm)r)z|Yr!Te-?5wb{>gCSf?)^f3Fp)~C
z$&5T0g{+!9mXVYyt;4%xy)i*a4N2J~l0euhJZKH_pwJob5?B+v!kW^9RUO2&XK{6J
zm*=iuw5Mq=;S-oSTkMDWWmpIpfTsj7FK7l%z~ZLBDs*-35{@SajwchUoG1f1;xyeU
zjVdxVJ<^zz(p1Ux4oGTh2;7rdb10e7q!^_#eTO%l%r`nv-Qo(3!pzK8ptD5Pgf$s!
z4-(QCcr1lJ4hj#!UpWL~lKjONoN1eGJ9W<j@wp1z+54`0{WJb){{rs}*4=cteKiZU
zBoLa}GQH*0Hr>BwzW#3gp3%AbF+DK0;2?pux5C)&7Hj=4CYKaHbbWe#r`Pctud6fc
zSsBo;gVOf_=oE%n@GAosJ;H6kwS55}AOC&TSr>7Zk|7ewB@h)If(Y<RsD->#rOCjT
z&JD;k%OahGg|MjAbR`HfpcU0t1R)y_YgR%b3owN3N02q?Tr!gYtDHtVTQ#eQgKmV4
zrkg+%VU)~d=w>KZL8gHr%tnkgb@~)&AzqYiASTK6O02A|fQk#D;sL1Ck-CPNiRp<`
zKfcxAbFD^HtiFL_NkLz7dd(Kc$F5FSH6qa5FcUWn8Hk5U25y2}C`CMK$*4_~Oqp5?
zun9T-16X2Hh2eq->E~du${J8|yydd{o8k)2h)cr(mm8@L!FUx2I$JhE7r(_&883o)
zn+07oaZ`-emy8pZp_|p;2NQQ07>{ObR{t=nA?@_Hpj@u|1TBLxQ#q7%)=vr=I)|vB
z|Dx0ph7wC$G#+uXlQEqcU^Hz2HTo!s2u~k_#B^JyS9QFN)fS<6zR!Sq4F1Yi5R>F@
zLE&8ET=0R(o%0_5$(<)At~Ia!(zAZP_Ab3G_SwK^&+7Y+UQ1~D(Gh)QRBsuZt1al>
z0$2fWbP+lrEA#wkQ1gB8T;yAD4*v_=9r5wCMS}0>J{UjH{!m1)e2C?9tfOEd+B+3I
zqL08p%PCY}MWGbl7ibq~?t_1EhbtJwyVjH#q&Gf(&HKdlP`$;Svq>W)+;V}&Edl~!
z2G^RUMMj`?Pdh$!b$Y7_gw$XJ!sB0=Kq7a%73S<O(+E6ykz7CqN5rY^fX3fNh6+~=
z?#|j8baCM4Y@Y^5wgOs>VrSp!AS>Dkf1P~uzjKZlI9uWNWgN8>UFaV%hQ)ybxZ@Tb
zI7SBrS1DM2vALQBL&*^-LFG|}?gE`&_$v_*Km<?A*>!WCd(UnB(z6cWzd%T9oCl1D
zx_QX^KmzFRr=6hv{|NfuR=J238PHy(f`M02jU7DPN0!wuC(bgJUA^50dx|cLU9XpE
z?Cw0YC%#K4v2ojN5ne>79++{HB6OGYvAK$mB79;<LtrQ=bPuRw1gBuTD!1mZASGNE
z(5?Bvtu>b1nnlfK(7Q!I+(5BJ(S}yx0OM=MNc@hnG9&R4PeT=aEV`_>Dq?QN%Wml6
zJ9vq~awRoc*vUfe%ixMIw!Wf&Xk9q%ql;>IIReM>9`invp!Yt;s1Skcb|bj7;vLm)
zBD8ByKpVk0X-q{U7GW#+XS`<d_g4_;pMWs;BjW^g1-=jbn{mo`8GIJcw!O+x)s)Yq
z%2~Bl(=;b1%8C@N=^*?SaQR5paQpejxo~uH?;kv2=Bf66@#GgBz|zB950bU@hm5$r
zjJP0yxb=*<_*I8Wsz!6BGtR#JP`tE^x7eShmVMoUHJ}FV01;_2YZ8e84cbeI1ck#U
z6#cvDhQKhL8>CKrgE*jgg`@2#9!7!Tj_HCA1rlW!Z05ktP$E(BkqmmoBVxfi^Y!!@
zEOwzlTV?GfZpAx<Er!{*CGVE3ENU5P8$AG3X!w+qAQl`P$Nh<{{f7AdOxg|7{tXE&
zcu8&Z$zFD|ozU;!JQsXyvgUfzy{{iTlRll+h0Pa7=9;#B)zoD)b<H*H)b}3Mn~v(F
z;fB+-k=IF+e%Ja%;^O!lb@iNk@gYA~;1=($;Z`rUI=OLfvDU$@SqwP1ZhkQk;MUw+
L?c{oSHq-wAprF!S

literal 0
HcmV?d00001

diff --git a/airflow-greenplum/airflow/dags/kafka_to_greenplum.py b/airflow-greenplum/airflow/dags/kafka_to_greenplum.py
index a3e92f0..969f2b9 100644
--- a/airflow-greenplum/airflow/dags/kafka_to_greenplum.py
+++ b/airflow-greenplum/airflow/dags/kafka_to_greenplum.py
@@ -13,10 +13,11 @@ from psycopg2.extras import execute_values
 
 from helpers.greenplum import get_gp_conn
 
-KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "kafka:9092")
+KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "kafka:29092")
 TOPIC = os.getenv("KAFKA_TOPIC", "orders")
 BATCH_SIZE = int(os.getenv("KAFKA_BATCH_SIZE", "500"))
 POLL_TIMEOUT_S = int(os.getenv("KAFKA_POLL_TIMEOUT", "10"))
+MAX_EMPTY_POLLS = int(os.getenv("KAFKA_MAX_EMPTY_POLLS", "3"))
 def _create_table():
     ddl = """
     CREATE TABLE IF NOT EXISTS public.orders (
@@ -86,15 +87,13 @@ def _consume_and_load(max_messages=1000, timeout_s: Optional[int] = None):
     with get_gp_conn() as conn, conn.cursor() as cur:
         batch: List[Tuple] = []
         consumed = 0
-        while consumed < max_messages:
+        empty_polls = 0
+        while consumed < max_messages and empty_polls < MAX_EMPTY_POLLS:
             msg = consumer.poll(timeout_s)
             if msg is None:
-                # Нет новых сообщений — сбрасываем остаток батча и выходим
-                if batch:
-                    _flush_batch(cur, batch)
-                    conn.commit()
-                    batch.clear()
-                break
+                empty_polls += 1
+                continue
+            empty_polls = 0
             if msg.error():
                 raise KafkaException(msg.error())
 
@@ -102,7 +101,7 @@ def _consume_and_load(max_messages=1000, timeout_s: Optional[int] = None):
             batch.append(
                 (
                     int(data["order_id"]),
-                    data["order_ts"],
+                    datetime.fromisoformat(data["order_ts"]),
                     int(data["customer_id"]),
                     float(data["amount"]),
                 )