From 9c429caa565b9b485b691047ee7ec6f29ca138f8 Mon Sep 17 00:00:00 2001
From: Joshua Salako <salakojoshua1234@gmail.com>
Date: Fri, 2 May 2025 14:25:31 +0100
Subject: [PATCH] Implement streaming pipeline endpoint for batch processing -
 Added `/run/streaming-pipeline` endpoint to process data in batches from
 either a database or CSV file. - Introduced `BatchResponse` model for
 structured responses. - Updated README with new endpoint details, including
 parameters and example usage. - Enhanced error handling and logging during
 batch processing. - Ensured data preprocessing and NaN handling in analysis
 functions.

---
 README.md                                     |  32 +++
 .../__pycache__/api.cpython-311.pyc           | Bin 17046 -> 25064 bytes
 .../__pycache__/config.cpython-311.pyc        | Bin 2697 -> 3691 bytes
 .../__pycache__/data_loader.cpython-311.pyc   | Bin 10181 -> 10300 bytes
 .../__pycache__/main.cpython-311.pyc          | Bin 9871 -> 9871 bytes
 .../salary_earner_analyzer.cpython-311.pyc    | Bin 8276 -> 8880 bytes
 .../salary_predictor.cpython-311.pyc          | Bin 10835 -> 11520 bytes
 salary_analytics/api.py                       | 183 +++++++++++++++++-
 salary_analytics/data_loader.py               |   2 +
 salary_analytics/salary_earner_analyzer.py    |  40 +++-
 10 files changed, 246 insertions(+), 11 deletions(-)

diff --git a/README.md b/README.md
index cbc96a5..91850a6 100644
--- a/README.md
+++ b/README.md
@@ -119,6 +119,32 @@ uvicorn salary_analytics.api:app --reload
 
 6. **Pipeline**
    - `POST /run/pipeline`: Run complete pipeline
+   - `POST /run/streaming-pipeline`: Run pipeline in batches
+     - Parameters:
+       - `source`: Data source ('db' or 'csv')
+       - `file`: CSV file (required if source is 'csv')
+       - `batch_size`: Number of rows to process in each batch (default: 10000)
+     - Example:
+       ```bash
+       # Run streaming pipeline from database
+       curl -X POST "http://localhost:8000/run/streaming-pipeline?source=db&batch_size=5000"
+       
+       # Run streaming pipeline from CSV
+       curl -X POST "http://localhost:8000/run/streaming-pipeline?source=csv&batch_size=5000" -F "file=@path/to/your/file.csv"
+       ```
+     - Response:
+       ```json
+       [
+         {
+           "batch_number": 1,
+           "total_batches": 10,
+           "processed_rows": 5000,
+           "results_path": "/app/output/csv/batch_results_20240315_123456/batch_1_results.csv",
+           "message": "Successfully processed batch 1 of 10"
+         },
+         // ... more batch responses ...
+       ]
+       ```
 
 ### Workflow
 
@@ -127,6 +153,12 @@ uvicorn salary_analytics.api:app --reload
 3. Run any of the analysis endpoints
 4. Generate and download reports as needed
 
+For large datasets, use the streaming pipeline endpoint:
+1. Start the API server
+2. Run the streaming pipeline with appropriate batch size
+3. Monitor batch processing progress
+4. Access results in the batch results directory
+
 Note: All analysis endpoints require data to be loaded first. If you try to run any analysis without loading data, you'll receive a 400 error with a message to load data first.
 
 ## Docker Deployment
diff --git a/salary_analytics/__pycache__/api.cpython-311.pyc b/salary_analytics/__pycache__/api.cpython-311.pyc
index 939b1f5484e841baf58e16cc5c546cab4e347314..374ad32cc2419cd20d985fa180ab38af80261fac 100644
GIT binary patch
delta 9014
zcmdTqZB!iBbu;_TvcN73yTixA0L#a+fIxu22#}D3BrJp!K*+{YH*02)R$6w~vx6nk
zC9A}$O>D=(9wk=%QIj@x6$hUvhosdXH%^`ASjQh{mb10CQ>#%<PfwHdw8h8HsZMfE
z(|g~tzyfxX{^&U!-rT+KzWd(0@6Nq%?tR18|AhYeTXfxD=H!?uxSsoK=IrAKDe9lG
zlD<r<$j3VUvvp&!DeJhErg)k+25jRt@|!=NPksx=3&<}sUPylJ<97Hp1sqe3aR=#>
z6=0`|#)|+q^Vxyosgm&$f?EQmQ_gYcRM~hLspkaBr?_#B;JJZ{smk$6g69QdRa35U
z7fop>q2OF~T1LQY(l~Ejw7sgKsPDj?x>mVs({-WlRT|piPF>?}7)`_H^94M!Soo?I
z+TdPos~74QJn0%m$sf|@m|*7ZXK3C*AYhB^t9s}Jcj_V&mb?m;@1#l8cmr=7FNeDk
z?h3f8G}IxAFS<hU#e!wiq)nu)1lmfswC&)toL8vvW{|5zem1*M_G&8iJE7K!wOTH7
zZG6QQYLg&sYrK`4c&oPXwgc}j99QoWnii{NPFtbxNYgyENXyhUU$<q9&NY7bHvIKl
z`0q>ew|KVUZxD2o8ou$0Zqur|(!D%Q+w|J8WtQE*Pc+upyh#f!LO0B?bIS}pYcsTN
zGeg^!8G6(Ft?ip;=;b?vZoX5{PwM#luIM+dWKX)^K$l<@b`-6d&=y_qNsq&K3wtLG
ze9xA#`gr3A<=!h_)4kH5RZYjjl0Ovm1ytPu|AeIK2K^Cf$W5yz-X{r?e@XycCkYoN
zc|c!LqERi!PK+KuG3q_uH+pzPHIMe~8|?QE^&RbBx}+b`SY05Rn2o@Kpj!UWc+^~p
zD1u#bt7({POjjoWS-aBy_3=SjH8^8}PSEq(ZfGW$j$j6$>DMp=Vq`TYHD{ksnm9Ww
zM8ct9M3`}<F>Wfv3juC2BytZU2NxDY6G9~751wgKErjF^PES27h-!`$l6(O#p$V~w
zn!Cy>@Ln-=A);DDAu=71BHpl1IxAMe+(T}oSOdQz?jvFy0yh9v>kmpI?hs-<f(8U+
z(M^aorvZu|IR@bJhQFk)NzpYW_vaTXIxKNrVY0xk=&;0f_9Wv_bXej#M{*-ij`i~3
z<-sq)eR(i7S@Y6UCJSAgIlF3`I4ew?_rlZlc>^IIFYsayh;Tsuo2*W{ap}LZv^u(8
z{!LD85elp3L?oXmO^3b0qXI1IexMC5?auvxrmN(9YprW95X3$No95App^zjFLCfdm
z)7ECXQGUky{o143a9Yj^zJPSrO9T^#fn!qk+jf93-?05eV?afVE?Hn|_)b8F+}T86
zaWB%z%)0?o8F4!3Juh6k5E6N>4-EQJ#2-<$0U<~%PwYqbg9yko(E;{tjY!OO0PqOD
z8aPgDlK+z_)SrS%K>lxM-BL?oSZh21)tQ$4@_(>J@-wVieuvF9i~&tNDgTu9I3Gat
z(`1uti(H;@<RD(KjTquX(7HbVqebQV&jTfj^UHr*^cH=3>G9&nH1rAC?A(=}_n@=>
zcg*`0XVo7x?>{=r^yrXJ!@U0nbN;f-s-;iLzp1GF9TRm`x=xa{kyUQ9K(|)pVPMrL
z3gM6_MMR=-^mh(5j!o7xLQoK4$ML3lf<S(0=?9e^TClW#uP)yxgn5ev$nF|f@24r)
zO{ZqWxBKdR7D@E^gI=;ZL_`r9ZY(`j^S;h-22gQQ_IgI_=K)o71OD?u;F4GHi6CGE
zn%?#V=}x($;WZCF_aS%AW?}Mi2x4hyWU3|<5ocvfW62Ss8niyQPK;oadD0&c$aWGD
ztFaeR4E6%WtV){4hb{!sQCBr1UV-KxE{ToZx`H*sn<+o;4+{Z*P>_GbvUFHhTS_O<
ztv7p#4ZUP?IuP)NS39gry1_t+kQ&t(nV!JSG9_2qLvYx~)8J=Ir*{5KpAW7y2P9QH
z5qVVX0PNTDvCe8UI)(TZQvE__W9zSVP0;>rjcR~BS&#(P5SgBw^j{ouYs4=jD^UUt
z1dogm9+3a3vxt5|zSrq797m2S*?r&SapLh=L@T(FiI<^CHB1LV1Ls9zhPbziR|#1P
z1>l{Q+5q+<d2JVf?VHT<r2NzS>Z+f>Hu40FfZ-E>i<qbQA~qrTWT~NRMze3mu>%yj
z14b2mQ}8NjOl$KOR(KVt=G1#7;-3*F(9F}96_Cf^{@7rVztcLYk+kyX>}>w5HmZqg
zy{KPuNk;jLW=6hMoBLd|!)DBk6lQhuZ`}@g&QU1;YkQ7tsW;PE5REap5|i!^FGZ8w
z%&{>(E2?{udRdbhkIm|(oORZ7__8KfRL`3ivom>V>I^k&;4QO8KIaNGYl>35{yY(M
zULzio)&mjh+}fCYE}yq(g>ect1l;Y^Y*v(B$d!Gr%KXPD-uB1ZG3t5k<GNXMG;1Lb
zS5Ol*XD*edqFLuqXXKjA=56>VztQFys7g~Q>KytTxuV%o&0>D$0uE5~^ywF7Em0$1
zFasja8l{>vKSeDt7|3dwC}`Zxuo^0Aidr&-n6*eAd`kA&oTw$5<F%)IXHn6d1qZCh
zr2~XbBSmSX7?-M<%}qb92GDO|)c7LB+h5jZ3OSn>&E-w&%3LTxeH2IY_}oQDCM)Q~
zuuzK4y%KuC>IKVVC3N_7ebXvns3M&EWzAx7Dlw5myX<u|Xy<8b6(#kiJ5rR35@WO0
z3@ve}rQIb&_)=;AI&ajPsVta4mq}Td_mBP%d3h)Qdrr4;Ju#Yk5KL><`UNYj>uL(S
zKwqFPY9FL7&{<T5;p6LqT+!^!GS6#*FN0RrOd720Gq}^^N4468DtziYBQ|L*l>~1r
z<I5Lyna}Y{bX0rJm6jmY%??gtla8c`(2<h{A-{aFzSuCEAGIyi;&acB=4Xn#P=|Ax
zW((H*qi%1sAgY;xfxzg^5K3rfz-MO}@Z-X0VYC3>Y!p0t@x#w@ix{{CSZhU;+3vCO
zl_*>K0=}xK2B0fi2;7EQyR^|o-D$q4edDq-RRT_GIfri^a*3MKc{+II|CLJ~$TaaN
z^ZU4DTEQC)xrU~KPi=9_FZ>_4<!8(MzjMp+b+-(fHajKQqlRqE8J2nD!>D0|+M<D@
z+i4)LI&({s<GXg`m2INsbySp)jbVK|aM$?GftQ-`H7}2~)Ms=4U@A!zBC^<VwFev4
z-{f2!WOeQ?Zir-yTxgQRtdNsJ>68+?3qD9ONgtW+EQw(<{m{ORnd#{+4vD1uYO)$3
z+39-E^mqbKL=&F|e;_fH_zYreF_rjLsLte6KUh6g&3mg4S0Ak&5hsDB=BA_;kF6%G
zqnOVN7hzs+=)8!p?}-iD6NTYU0o)%a0X$D_+`3RfsSDoxGz1kz74VC})u(c4kjq5z
z8sv(~KPqP{w7ibj^9J7dl5WwIxp~b}QEEYp8(0icIGM<r={XKbKq^h=eB8v@>EL<P
z7e;#A1;2Eblg<iU#5W~yfzTQM1m}xz2cT^bvstl8&6UK9UXr$AmMf?b>tFN=K|WP8
zVNJRe7SwFQH~lcFSp2b|U-AP#?|t}^nmsWckwQ~~==Jk%ovPtm;NA&$E8J}monmN-
z4PhR}36Y`kutYV8LJ%Z_2v-b+gFaBao-81OicBXv;kETa(hIAbX@+Pr1~RXPlWW0w
zQ+FUV;R{51n>KgZ5938Hz?he*Um2)c_qT#KPAI#NB-#h#HAfRQN9X#L{K~kc>W-`C
zTSGV6Z<Zxm564{t3D>|}|9n&2QgN3ldVcV^!DO*3SzMbeu1m&>Yge-L_4XAC!2IAx
zl)ljM34*!)6$@2ZjNH!-T^*Y1zgxi0rOp@*it`|X_$&ZCGs;RW6~bN-!?qdQ@wLO2
z)zcT%kTT5~3#2pTGzBvP)Dq~Q7cTzdw{XokM^bqQCMuWK-C=bMw_a?)S$866MbL%-
z-#0Ukeo=&@p7leX)jl(tE>5LHOqgbz2Yrxs!MioYO@x9$VM0QbkAVq2?1QtT!jb;L
z{{5re{$nSGM(gU`+`-{vN4Xgt+Tcv?YGeH>)yNGF932?tW=!0%g9k_Y0duBgfZbs2
zfpY2mpo?>RL`?8RjMh~%QJ5DK1u>S36m2kcK{Y|gNi-*t*od>(pgR}x2UR2Q7ty{%
z40=`5l<&L%l!*9ssEW@cxQ5^x2wp(&Jpif+?27kdYPJ_?eF$s_NS5><VqPi+@L@y)
z08|4QHFQu7zHk_PCv|AGU4s7-avN6_sTzdf8OTD#JRAu36xAqP6egxgt_POv6IX?a
zJQ7g#!c<tg<aVTzu-KE>=g+WDHoAK1bV^jSP!ZtHQ(@JbGM>=16rPs6uy8dCtcaZW
zMN&5LS2)I>B6tPC0)ihPz*t$F0-)xvZjKud^Dyg+sXAN*xPE9lD0v^57Q{=cIdvw8
zJ3X;C)exGTj0lpdC+!gjbi~)hHdpFk!;T5}njNe;w)pqZu?l_o9PGY7hri2uYNhx=
zjoj0<F9sg%sCYJTH87W*tnkcbCro8PDs)~Y4tSgKEHj?#$Kp(Pg6W>ix>MqOspFOI
z7rI~9EE>O?_gdZ?mN!o)I!EGlqlvmvWlU7ggb|L3@sdcQB%+wuJ0|nf*$U&jE-9wA
zxT!5+Y6IjhW1q`PI@ppq%Ze^Ht7wH<0~}?^pFi>Bo-fN!bp3g(A*cRdoaHY)biM4&
zF2y+%cMc_-Ln~CRt>hyF^ZiMNTj+~3)$@aQnBwP;Ja<Il_T3l+042`sPcZuxX8($T
zayBZZ4cAY^OIs7At@Hh`NUkGZ-kB)xoIjkbtero6^~j2@w6NqZ%iU(%m)Z6>+mT>n
z9SYl#MADvRc2At`OR#+k+eaw9%WQ9)-J4+dD(qgEp~`c+a$vb~AYOSSQF-K=5vFKr
zUOf5TQ?H#;y6#s7L&^ysLQ>+5;Y4FtVe6A@`9e;DZAwk5>^iI*KcV34V+nRliLqmM
z*^1k2-7;JEx-;o+Pu4fw@ibkZ{&v@5SJJ&Z=^B1>P;m`IS;;ok7k@+n_yhr9MJ0e0
zm8>8j)qB6nrAn(;C{rO2)8GzU_KM{Ni&A~^W)uLF_ae&3=&eWJeN?HAVLi@1kYFEB
z*awoIA&MvI8oL=#Tw_q)n^1;FZcV;BnW}!aWJJ78ASLa+d-q~1dXN*4l|AsZe>F-|
zwt}l!*P7y1cf#sctZqQ=eNB@=0xEVeP)zylf|})mns`B7qM&YtqH|8uckER+v`Y6w
ziv3jFekx%<rC3h^YrZpKtGaFTEZaPBTVul3xI$@i_R=?E%$>4IrRvZdT>zlO%MK^X
z4$o&NofWT~dg0V{Pu#gH;oNoG*|Y5Ii97csoO|ZYD|+mk8X*NgBBOmm;70nV;V3_0
zj%~wXqRQ&7*T<c0Hy(*QJ8zE1OApN*O%~ec4&KdYuXQcwSI6_Klh)D&I&Q6)x7@MX
zpU-|STPgkA4GRD$Z-Oh}Z`?YZunsHMVNgKPgY@gg@Q9(@yg&7;97p>@%Dny#Tl`Y?
zLjU#M2~SU)?M<+~^LltUU)>E_wH5yZQ-njf6}IkrVVvEOV0X;x!Td{G<0Wm0lD2tM
zvaD*}bTu11%;vbul-_1qmYJ3~)0$vf6{a<bl-<kB?l{wvV0sj$hfunfnOIk9>rt3)
zn4!G(c3J;&S%19jP@?S6{2`d4zG2bv-O|@em0bsw!{?RHokmDX+#N``0}4}<WJ<0b
zO)&MTNtK-k-tJd$^5FzCtT4lO8Ru=rwamEw;&igQC0WyYr?&oj`!{F4HIuCFh$T7S
zn-+!hVHx(O5U}SOJD6yV9sG-67`_+^9bm<)06mSZ04K8nF%58og2+ax7`fRE0Lpu<
zZ$ENt*Sot?wKy}HU`7>YG`S{HY^zMQNpARNgTf6%d9Uhi->sT=Yf`n}L#}lR*D)}*
zd*Eg^2mUgY&2g<hZgtHYR(v$&C@4x>)Cx~i)*{%^@7>8}<Q>H<+w+5wo3`(beSb_T
z?12(D^(IWcim6xrYE?}v0dJiv`<uqx)LT0{4zz23?6v^@<96MF`;0%nPm6drjd%~@
ze@FL~44SE*IQN?dYBjg2HBh-#Yd+Gfztz&$cEqE9r`iCdcRYF^z0<5kyv+_B-!<C@
zIqkb#9}SiFD)tODXy4~xrWo~pgKntV_<l1E|9&ftc-y`_sKj;qvq!B|A`iPHtmZLR
zn<(yU8!OO$kgEsM2L*Z{eZX1~cbgxev>&wgnNIH3-rlW;3LGHgvGEXmY9PPjH=*X<
zm--xMob*Bb5cZ4m54*!LJk&|WRr>%_jgUh?$Z|$CgndEY7a<W-DxM+%nHeG#UjV*M
zO<f{UjjEf1F9#%iQVkF=O#07|V2ngN;yXAv3GJ*AjCL-0A#ms6!)izr3voKT{Iwnj
zZI@r~sf^)4B#FpWEks#ru8-%vQ+}Qg2p4>!kP2BygjR`+%?R*_Pc?;t8$QRW+A~7x
zLmT;ars~2Wh&N6l_vaBzAi%FF<TFR=>lc1Q5IY`6NRGDK5yOyV(^tG-o4Vl>$#g*I
z6@Lk>5U)@XMSisRLTRl+eQJ@ldzy!mlvBwpNvcH2EJ@0(WR@h=plnx`PVOn9>ulg3
z^ky(W0A5eqz}M4F)}`RyVol%_{Lh8{3HD2@Vy?Pz;HERtcIYx2rw=FS!z&s=LwB#h
kzj)By*9%g=H@i~5%J}INvJrlyr8EQdrk*JV^0D=Q02Dj?*#H0l

delta 2719
zcmb7GO>7%Q6yDv~j+;1%<21ic>UCl_b`s~O%}+@}_$j3ov}tG>pl*V@^-K~sYp<Dg
z+9tFF)RYS+G=c*hiiDH`&CL}jE>#iY1mS|p3ZzP0Dtaj@Cn_P{8^>wlq^cP2ynQq8
z&Ad0?y!Y1kZp)usl|$Fd%1R_?kM35@o(fCSA2`Uj;&tYsZu=<o`jRXWnH1~ps9U@%
zqm|<AiF(AlD!N;|tE1J#Np|Tq$=YZwtZ|YOy)IcFtrvW$-jMW0y@Ge?jmf5Hli<tr
zJxO2GCrgS%J?GTC3&8u=`3Z@XuU4!ml5_)luC1##=f|{VMFu(aT#E)^JB7GOCGo6Q
ztrUR_`noJggP-r^J*~UX(Gs$IMkdvQ0=HVTVh2^|xhBliuE5|%eoBhAlHzCs^bqt$
z=zA3DutaK?B~nLS1*XEJep!mP(I8LDH6hZl)l1@CmI^lS0N*0uTXkz=fo{__-Oju&
zZ&8}<T`=m#mF+4Gull#*MVhy*?%B|9*+D<BO@Cir-yTGL{+r^r@6XTmwC=D+h}xzV
z5?;0yoUu2*s=aN8Rqfj>_kq5^B+^lUNrd);h0bjj1~x2o?O>sMn}q{;{l1<83;kq2
z?I*p|K5Zj?%l3kkJVT1drSJgnwqNromh*TzlQ2@6ZrNT+#51G(k=<V*Tdw01C&wmE
z#>QSad1O30;22ljDnt`2MdCu@=bt){mFz(lNgw~qdBOon3B&BI5|`W>LS9U?0cTal
zXK8#cMzo9;(+!PK!uG-Z5dXXMfE><7Ttzk+0^HqNFVxG-jK(ta=@>mvQyKOgs2<F|
z?fylUef)^0HPDB<_aiB=V`GMqVFy9-2Irn`In2NF+-!Yu2Tj*3)%474OgP34gT@rU
zR@DX0{8aV5;_%@Tm48;Vw=M$gXt-23%pP0r=D*hX4hskdQF#c-UL-akUnxvQegue#
zML<=VI9i1Ua4#&<*c@HBU@*d717V!6c|+Nm+H{fp3cu4Z;CL0+9_7{E_NHUVKGBCM
zZ=n4+YCOpiHU`oS5BRFrXP>|suy|iSk^Q#uiUKyD_x0y(Uh=g)#pZ2a%m1=jsqV4A
z2{y0bDfy&&8|<oncaeOQ_XL`sVk{j942!dgQ|^FtxQLyC&8=ca(+10!Imoegs)#vX
z3)I--Fyh(!fq|lD@nRS24>4viHIY)22BEr|VOk=UNX-nYY!loi+pjzp3XD7n>9CwL
zjU|2kvRz{1m@s0p84z60-U|I_b4&uuPV<i=<27f1waWFx9Mu<MRAVX1Oc4DWNy!nO
z?EHEOJsJ&{6*^OaTZo{;c(WA4WDFSxy_hfo1}6@jXEl?iv?OH$ISkLRjm@H{WICbK
zSX#@>nhZZ&Ip=XNz%3uF#JJ!|VZz38=&}Wnf0O;PYuM(<A;tgPACxI~_cktJMsH3K
z7+RHVJ{6mu*Y#LBk*0bAPPgInrkE9uDORyLACFUW{SG&!4SNUoMsf$b`4GrG_`N4R
zaFpAJe|OgZ?DS@D^!{M?gcWufq*hVfJkL6STjW;<gC&?i?0qr3IT$&%XzK)dSg{;%
z{WL=@hdDnzowzs(%il#+fe3B{w<;#^w2M_QYXBNiGlq(ws94X~CEhvI?l^?2d^|QZ
zJ=P7ZwM(ET6y|a4m`}m!=a>lN9#mcy(-}jD>&jh^C@umXfk$!9@)EBe4h1jak`W*#
z{@1}Pf+k7|`w&G)?q|;qFDkBe2u8z$EXV_=FKg%k8vtSz>&DCs1iIx&Ybm0cmXapy
z1C-d3kOHh6Sx4MRO($j;MwnIlxMnlv#u>M%=^9&zr5Ppg(J&aR<lf0f*~1S^HaqYS
z4(sD*ChM0F2PTdxQU=4ravG^kMcpczp}BG`DzIfs8-N8?L^gp0s|Lg56}gkEt{8#2
za1iI-`T_XIISk1*uhS8>vI`V37tK2Ep1M%uUz2ut_!m>%fxA+8ZR>MKQrEUVS^r!0
Ta+bvElya#<9)5sCeBk~A5G_r7

diff --git a/salary_analytics/__pycache__/config.cpython-311.pyc b/salary_analytics/__pycache__/config.cpython-311.pyc
index 66e50b5fc9437f03259c9fea33cd4c4f05aea6e3..7495a52e8ce1108b6795a7f3dc130852bbb2967d 100644
GIT binary patch
delta 1755
zcma))&rj1}7{}kXY+cuu4aV57Dk2)Qs6-*+L1F@dC_31}z#qLdW$!Bt$GWV;D2W*+
z9@LPisfh-cuy{}ty?OEA#s6SSc4*$C9=VX{k%R9xhIEUvw|(2U&-e3u-=|Oe)VnR+
z`5=l;K*rdvXXyu7{j5GThPb!*wc7!}YXE@^AO-0ZB~#?ir05k!W+*@d#Jnfyr$!#2
z0HT&jPJI+@Mi=FA9Uik+IR+s1W%0DjVX@$<k}*K_xSXx=0f3y)_MRpskXaux{}0aI
z#0jljzUs#|aUyw5YPqxFPYB0n&dStUaggELV1!mC-(n=TLVJv>Q4n-O-dawr;J><X
zh%WBy`904jgHGuB?*?wzWz}JK!6v(*NA6k#u;(2mcSA4ig+AzqfkKDuv39=kA^;d6
zit7hRfI+LDm9PF0H7e8=CAC5E@JL(K(YC0*HYgDuYm4e{i#mQ7)d!FW2dor4QE<vV
zkdA}!N#iOro)1%)PiZPtpq4}Fd)Th6<}#}}Y|EsyoQ|1<PGTz>L@9launu%Tspk~!
zE@mgjmHGJOEOzQ@O3gkXCTYx*U~DWNzZspKs1Y+`)04GDEIK=fZRgKj7#^xUUya7+
z*8K3xcP7u|*L(>rtrKIE&MCFc2A8#kRB~Z0kWBwe@*fcNgqlLxMpDp{WYNM->{&`K
zE-6UOrcqXbD5oZkc{ZEJ?gM_c8gRG8rylD%K-HbG$AZ;^VJ~JC($j>ZU|Q3$z24p|
zLw<bRkx_F?n1#t~T3tbyRTuP(vCUqVxasJ`WJH;mn#GP4^)4a=NnJ$MFJlZj-tsp{
zIz^_wK^vw+>NCO7lJ(rN14>xuABm5}b<z0k$aqbmbK}bM$g{|sVDVO27_10`>zo0(
zE5<|4cWcK1#NH?48&fZmMX9cu?!XKB75C=sN2%-{sklc>!P`Xa2Yh8AToJ<iM5+H$
zSs1MdqsBMxD>D7T0=p$_3dP%{b5mviwTl0m8I%gAw$5yxDLyC-O_zf+mEepS2o-!=
y(xy~Qmd;-<2O^a~#0(uNT-+Mn9R0|ZhNI<BtP+atczHLy1IT4vs2<rl(u+U8V~LUg

delta 761
zcma)(Jxjw-6o$`DnlztHYa6XnP!Yu7VypfEhw9`~LC6w{C-~Kx(zK$R2o8?r4hnV<
z2L*K##8L1sIE1W%i%Y>jU?SF1i?ru(FZaIpIWI5VkIbo_zfn~Qf@(Z!TV(;@Exfc#
z{5gDPSb#G?FaZfNC1Mhr>@aQYlBHRa)S#~x2{{&QuJJsp@m+W{e3}BANA@}}zWX&d
zg<g)TB&CVHx&gL;+@E=g34_V)umQo}N9hEkYaLB&YwI0N3bk%80w{d9h!V<8&Qwq_
z)lES4ikLBsp@wlx;7~JSYQ5<u!Im)D?f4Is8i>*dphO%Vh#Kifje{*>x|_n$CU3@(
zDS$CXXSk&WpLgoDO1<WDm6B6)eRjj$Uo(g=*azEg&2n~q$t{+O)x(lg-1E8Y+~R!h
z;hj5X8H1;<;)!M?{HSH2vte02<GA!iDu_b4xNGBf)ulQ4LS6|zk_5}WCDfBMG^teR
nxw4#S380R8@r<XWJtgTW({xpR(pzdY&a@!dj+C&g`C#-5zRHL^

diff --git a/salary_analytics/__pycache__/data_loader.cpython-311.pyc b/salary_analytics/__pycache__/data_loader.cpython-311.pyc
index c99c113d6995d2ac6f29b069b0f52f7aa3f6ccd4..a7479801a858f33498c153c3136afc304250db11 100644
GIT binary patch
delta 964
zcmZ`%T}V@582-MU?aZH@pPlW@ZMq-bkF|clY|C^@mh>}PX=z1~*^iSP<n%Z%f)#cp
zb`i##gb{cyLN6>R20;`=t-weE?@GK0%DZ4fUf4x-zT>76^}%`0^Pcy4f6jBxyVjR&
z%rwK4A?1iK%va7lWtPhIPqYZ#L!}_IUT_*d;!*f+ctugN0lsL>;I^~$bt<Eg86_0q
zEhBKZ(hTXac9JZ7V_YR%AtYUle7d@rMks@l97Nd>u_!@FbpfJ~jfEK==woX*Gtm|+
zlL8FTRdAK{=tQ;0V{W@fHWvzNgw5J^<U#2G{N{pqM>fdMVq5{gjBRwcl<M!h2yx+I
zX;6dEFv@HFuqyP^9OKI)4pOEe&gPsn+&4_H57;r?C*G1>GBsnn1-*70Y&dK|vzD@$
zNX@zfF`jzQ>tZ4_ixoJNJuN=d;4xsWGn8KDHk{IDv@)+cHE-P=`kEGtuY*w1SmYUP
zcgZ7hO4(>exz<hwbqNIuQM=b@N!J!rs)vH&p$$fDePU6DlIEg977H81$~UqW)xR-&
zEf!$Wu6SqN?A^Lf^==pJtPJ8HbkTL$@ycn8D<R_O^d40PNaSNn#|VT8v=V3|aGbyi
zc<$IHb|`d#PWAK*CejPKfuV$SO*#pyj{WpmB{O&dqRzO|{<d=&+hEw0``dLm;11CB
zHpafv06$Hx>|M73?+v3jY_zH+MCAg8BVm*aX<Higxkifc2qt{rp@<TE2$#j1s5v?i
z4^eX=UA!s#xw--4^B{P7y`5x9Cjk$EQwnUQBwo2K(1WQV>2gA9he=Nzp2*I7ZelzI
z?Y?@YcElItiC9t_Tp}z1bH2E1i_V%y>(GHCBwmLbu*W~n_Yk={l>F*aFD&}4xF1&h
Xcbd;>a77;d8<*ARKmL=IOiHgmNWk7{

delta 861
zcmZ`%O=uHA6rQ)6Y_-Ym=BJzHr%h@$|D|ACTeZd(Eoe)ljcq+B29Y+UhL)IZA{2^0
zc+sOgMLZdc&_hAcKtU*o1|e7vo_i559<)M=6-p1{%$B4V9hh&v`QG<t_PyD)<cCpl
zSri)y9rwQ2XI{S&znS1ZZku+XfIrP|d9G;2uLcD-MVovxFEZoa$cQLPK9I0O>X3ak
zm5@b}k?^_Gfw!d`jc!SHuwc%86ytKpB-MNwE9Qh2JRz&H!ZK=l(MG*~d?N>N&eJT{
z^>Lz-d<4Um9%wJ6EENC;hHOc^Wm`HJ;Rv}ys)is&?KcET3|5JR&+G-x+FVkb2B3>0
z?^@3UhB#8;8i#?em}5h6z7jW5J_)s(>DoXiPv@_R(j#S+gGC&1uJDGUjE7vq^5?v=
z@6A|n!Q}%hK6Q1-!F`?9V%4kv;mRr2nx3)OSh0!Lmon`~6ul42ZdLZM+`7-9_aW%n
z>qA}F`*0Z1(*xajRp>13dX@p)xYU~RvB%NQF^Drb&tQ~6lEDR3z3s|4r5i$adU`&c
z-!SG1X>CE9z)A0E;S!C^kK>9r0Y`Ahy9G}C?km$C;#ylw=yZzi-@T>HwstrkA@8MA
zErY#$Ic6S-kaDkKu&<%q$1{Es81FwDrJMCgV<OD2cf}HZ{*&LB2$z!nV*n|9pk5DL
zVc+Cspfb2fA<AnOx@-vZ*@8Bi*2d5g=!7Swp}<{$7x*#QO>_SQdo0YX=?7e6YzhZM
z318Ugtdcs6L7eUDu!t{1kE3a3*Dqy%yf(vtJ&nftyL%u$mz^riru(%l&W0ZjT;`y;
PO8$*oy1xCNsD9{QNW8%9

diff --git a/salary_analytics/__pycache__/main.cpython-311.pyc b/salary_analytics/__pycache__/main.cpython-311.pyc
index 4f1a3ec44893c33145e870944df2c8c6a9088771..6727b9c21e8b6ed5a662f056b4744f3815fa6e45 100644
GIT binary patch
delta 23
dcmeD8?f2zd&dbZi00c+Zie!9{+{o9c1^`dW2JZj>

delta 23
dcmeD8?f2zd&dbZi00eojgflv&Hu5#90RT)K25tZV

diff --git a/salary_analytics/__pycache__/salary_earner_analyzer.cpython-311.pyc b/salary_analytics/__pycache__/salary_earner_analyzer.cpython-311.pyc
index d72e45e7e093680f68d707f48184809dff91aaaa..828d0ba2db733ac67cd868e5c2069958194ed693 100644
GIT binary patch
delta 1775
zcmZ`(Pi)gx8258x$B7*~X+qj2O`L}QODRPRhSrkO0fUKg;Ln)Eq@l>fX-P?(=*5AS
zIjgvAl~!q4-^7^4A=`n0-~cLz-8XI1E^DetjXX`{$_0Y9-L|p!T#~I~`(*#}{J!t|
z<@@(u{ndL%#IM|L7lq}_H~tIn_uUsC(qIierVpd{e1l$_&Jf+E7@%rHSA9NoPms_Y
zCyEAr1-q8*h8+fUC%S6e5#rm%6g9`T*Q#R{=o0$g)(<%J$QB0J_Bl=5J}H*DSxR+&
z3Cd1`tGb9PL>bPzuA6|O?%;P*%ACQyNyT|IYyWP)&2%cMRKBF?8PybZIG-%)3e=OT
zqNhz($q~(MV6w&1`C?j+^H9JfrahG}ETBEiUYE(LFkjG=xC>>NquWh>CX?0EFs&xl
z8PlDD89kFyvdL6FTP%|5M5E{da|--{SawL*5GU8g$(op`iwQKwex3;QUXOpGE%7au
z>OBP3dJf`e$=wL_Zv;lx10(m`Yk`qk;6OcaV2N)^!8I-jeV7!6F|0P(Fb(#hzqkt6
zjy`fW!7wVgp3p;fr0_3?et3t*6}FKY&vi-x!Lq0`=nH<Q+kq*b%bXn@6auu)qhI;G
zfqz-sRHp3Ef(EOy=$x>NaTs>I`dApJt6aMg5)6^u-ZmMkb$6p@g2Zzgjomq{q|rlR
zJT}+q@EobBC0OZ5;1&7prO@M^2zK=(bduk5nUmTp!@0GukXAX|dCcGWbE@iT$HkUi
zh6|_kx_kJ=T%(4I4B$V3l--6~bzZazRVF36?+(4_ChDCWC`xsqXZ!#fcZj0OU$myZ
zEE#mwYejHU>^d&o;I>8#s0vvAaZ7exo$TnXwq*~VZ8x56ui>fs2xpiwysCItx``Wm
z2fv$A_81;K-5&IlI1wdvgAojm9$<(A)2=EDMM#XHi~;SI0^OE=Cov+We?*jV4u;X&
zQhCu1PvB2@lE5hfr!kl|C7X@Yu-{r=$X|~8O~)+EmkQ?>OeUQx=nHU!7~3raEq@u3
za+tPUMl%^LuWJgX&=yR4F0E)drp(NYNvk?MMy%LeTFodL9Ks)irae<6);VP!pT<HV
zt*NH-HAPpBLM4}mqu3_ig{C}%Zj!3WSW&<eDDPSJn}SxtHks7s=_8emu6Q$!y?9aF
zL8F4VZ+xdydP-ra34Ke)Dy636Uw*em0e5f1)AKt^1;f|h`Yd=Oxby}-k>MD+?H$cX
zeRV0;qHOM|h8(R36`?7Q-8x&B_if0N>+)nxPSoW@MZgZ8ZI!djiyP9wx-?LehU(JL
zhBUe^jn<^Gx->>8<VHAFedR`SBfNV(yt@|OQxET{xGJt@I8t$;KfQ_WUnKAH<;r4B
z8m>#jB(lLqY@i~ciLOSZ**&{vJxfBv7y5L1W%}0G*GDSTHQ%AS@6c+*7Xjcjn(~Jx
ztT`Y}JY&rOzOmwVNKQPAVK8}^#+O?%qp9h6t8;M%dWg1&R{WPhcGbCUf&pQ4I2dDK
z6f00Zc!ec(geCMO*b9b`GZc#(WP=<}c$9#3Q?hzKnJQj_9NuZO&Y&})?O+kjhmHe%
z_1jR2#@r*3UE3^1%eO=bEWQs=HZt{+j|sJ?4vg}QzeR0@;6A3OMRlNnevTZ$arMhG
lm_!HV{oq4XmV3Yox+X79pQC}RMg2EEwf2AeAFXxY{taObwA=sy

delta 1273
zcmZ`(OKcle6rDF;kL|G?J56nm$F)NOxXnijrEMTIh?0^f-Bc77RV8ccnI?AXPv49y
zr?R07Bm&AJa2G8ih%CxSLPACEwkuFui4CqSH5wM}0v4tcO#unPiub0qOH_D^J8$ke
z_uMn@&dimW_RE1U6vfMsb@yImu@?O%aEC=U{K_5*H#nQ4z{6EX8!WVf{V>f3;5|-7
z5`4z(QRObqFa@EzH>3q*D{zlX6PChj$n&UjmVcee$nY(HU6iq((BV5_FPrfID;^ds
z4^1qI$IuP9CO!@K#gAjb@Y>|+TR(U-93h}1>$zOHTC(#-^SY$5XbgUm*N_HP?_HFJ
zH+}cn0tc`8pZA?T$udlXYw$J)7yYA3&l<7GLCe1r$<Xod4H!bKYns^*T$gv0V4qay
zw;|#qB*YVdg#~3SRiJL&Rp^}o0mUhjqgVJ#84EXL%KujjU$=$et$S@5ehfVGgtvj&
zy7xR8$z$DyQdbOV3DI_qz_zKlh8{PtpLkHE=Z;|v4}1`OPLY>fRwGzvH~hrV@UVB?
z^Epxf%I!eAAZ87%TDD5}4_6@oYADXvLmR>FGf1I&h|JRmXG4?6MhMLb*tnLpY#rNK
zL$}S&Rg%mghd1<P%MtY=ot+b*cUdc6%9+_Sx@BicQD>AeJM`>2WN=KzgGb=cP{SW`
zqy=2AR!%MBLHN2adWbKYdWi_;PoH*J!^W==wj&lzBTv#?QLm9Um5Nz1aEc=GtGcZp
z#(L3oL<5&AB|Xyz(eSY1+7(=FJPUK-Rl({b1@KEaFO89n)kJV2lF&3c_<$jy<xi{}
zS*!jWOI&*6qxgD!Wo|Q(hRc!NyVYmEjJDN@+v>hs>b{nGpsgNQIl383UaGC{x*Z$6
z6&r2E#@n&+l{whoKLzjiZz`J+_2Tr0)2+x*J2KR~up^1kJbXH!zCc~!BP1MmPmLJP
z@K_@E4xcMmmK`59Ez{2CONLp)gCw63aER6{!0!X+(D7zr=cIr};BsQnOB6(_dW!Jj
z`^0&152Y=_ycR<#Ftx!<g(iYD9HfDMjgDxSvpMS=rtfxF`UG6jo+7`5wl<GWHdDzQ
zOB%eB8sFh6y0wc`z}2h6&D2X1ijdr5dLii(`nQ<vAwDETx0qfy3j5PD#8)G|3r)Z$
v=@;V{s3iU7F&)d_o~dHlsGczo;&<V<^xM-jEb?wK{~aH=`9J;#bw&RHUz;{`

diff --git a/salary_analytics/__pycache__/salary_predictor.cpython-311.pyc b/salary_analytics/__pycache__/salary_predictor.cpython-311.pyc
index 264f9fc22cfc2b32d97ea396338b998a4e1cbcf5..1d8ced8600065eabbeba0f94e7b75f160dad93fe 100644
GIT binary patch
delta 1744
zcma)6U2Gdg5Z-m-zxZt5*_RmG5MQ0Rj?=VFLQ4o#pfngK1VJDOh@n`n&-tr=(zAma
z<Um!B(ifzLDTGQqG!iLPNC;sG0hNk)uh3ao2o4G_yi^F0qVfVS?3@!P4h1o~`)+S$
z=iAwD=kDvLPRAYZ+3hWa*>``rXS{(G#~t+SO8DskDw{Gw))jYABAaDP#+?=8A|(bA
zBeL~1BDY*J5b^=LYZh;n4HG0{TY3MoB|^1kc_BMreu4%mbwbgwM4jc|4evJ57FadC
zMSXC<ylfI#rs9Mv=EJlVZknUCtvp~kPpPMT&HAi~`XR)x*|0Q4LO2W+-pvPb7)Izq
z@WCy9nWdM6A^IR(75eY<p!;rwI~m;I7(|SUA04}MF>XY#=uzl)rS|Q`;UGc)p__r$
zl$@Sc4(X$8@|0$r&B^IgN*{dc8lWO<xT16f?r~pWWvK3hwEJW6ktVi@G0&KR_Q0w)
z+&782&4rUnQ86Q{zTV?(6tOG13Mur;2x){V_|5wqqZ{+>r!T;w?-vv9DrJO<RV=^k
zrv^F(r|&uwev)}Lb0PU+dTMrBGo~{6LYi$>r<_8fMTBLDwEG!bw0(dou+)Bqdf=MX
zYb`YsvPw$iBCIuukPkdYzlW=VnRwC3vNo+m=Gd&775S|ZA##(q{?gb8l<dk!g>@_P
z@S)^!Hna^T@QoxH=8W)}tsO?B0PB#XqZO7Uk=k(NHkB;0sbOg|d@e<(1%8$moIe!p
zTlB0bE6nVPzJf=tTZts<(OpH==QJ?4{O?U|^X?SIhWYo_7XODT^%PbpcZsr(hut-!
z>=fqgx&#-TJilaTuk{Kq))esuw-n)+BKC&59ZV5lm&{L)205I&SHT$(0p1S2+N1HQ
ze6BECD5i78L^dy{Gn%l&Q%EH<Y2~Hf3D#Tk8y#Lj(&tFYoGjI1;OqFZS98tgc4XCf
z+zq}>rhc6X#ykDDq0&0u9Cs)um>3H7o-&TGgn;TQE(H8s6f1%a;Uq!<;dKN7!ObAz
z(A!^gL#0!MuR4=-5qd)>So^;dy5EePg-HYa7*dY@VIp?noOnL4-lHBIss@i%xZx@{
zta8KGIsTkj751zjQG1809S>EwXqAhqT=Y68sIKi-7~gwic$20o1lJ!``=Zs(hbvsH
z%EeSJw#n16{=C{BtA<7@+-Q{>Rk=}}XL}wDhD+?5{3N{ZP#f-J5J6W%*FfLk8uMKy
zivD+N#W=Q;4i0GTO2rNspP?J%_E^ipgh-oaP9)@fDv{7k^XxE_1!X_;YsMV=UoR??
z@I_aEo`Ijbjt$M_k7s6&>ym3GcF61(Q<Gj_c^#LpSM#Uk2s`_QO!@)kBNo8MF8m7f
ik+WHkN*b<jiKI4O8_mZI6pR1&$IV*o)_+XVYxx_=^oI5T

delta 1139
zcmaKrTS!xJ9LN9vZRav|o7<U|vo_~#9oN+qvx`_{f>IJjnG$2|EVDK*|IN^t(2ev#
zq4AR<>cQx0J}!C*Bzo~BC@LaO5BU(3`BKo6z4ZT`m0(nV&iU+jez))cwsW1Y583bA
zY&nFFw~rkszpP2ySJPBFa5yMxQp_25b-E}qk!GUFmxyM)XCh>VzuD2LXr^|e=A<7@
zXQH%}Qag;9MyNd_n(vsY6=uY%v<UV{qoTqKm>ph7wbTuQ+(Zj9t@0$Lg_(J4he$n8
zpMPtk6_b338V=nO3n!AgaH5ALlB$LIF;IyJAc{bCj`C(#oV#fQymD4=FGP1WqL^bL
zk3$uh2OWRbVq8GTMid%ckzfUy+Ynxa%AuRX-Q8@fVa3Bky3o_BMTgjCc<!pB3RsjT
z+6*nqbv})c$|kC2-0l`LUsGsRtBD3-rYNwb1B)$*0hVMEwlxF~a7(o@V^fZ!H-gX*
zQ80VDc)yTm13d;4o<-hnxY%Q&Eik$ELf|0J(WONAO!P!gw=P6teTgVvq|qsg^+Slu
zu*+M_n;iAl(E&(%pU^^>w-(8bW<su!6oD<yLYVd2Anj`u##3a<49=VaSoF2QyAm6f
zjISRhUMd^bCcm2>CeKjBl&qPPZm#ibXcmb1RhTPL;DJBlm`K`wM+nhG@RT-{<^J}h
zNC?lw7%6gU*`CVk)t0$)mvg7G%1aK1I3n{2X65s^ytG1I=JFnkmeWokuhc7|q2|Kp
zvWxJn{3(2>nA~S$r??`sBiaxtgahG3;90PK1VfA?@bAF#5iSnZ&J>ufJWGe5K5&|E
z>)pT(3A-fTQ1}#JaAVy9%&959$$P7UH3e8_+#^HSkTbg8<bi3zjBAQp$Z!V?*x+_m
zq`6!;Lcfx~Gh2lPs#}*rA+0YG3h82hxK|4&*m~~Qh2Ft<|1j%-{Xri+5Aoo!nn+(S
zzb4t`7-CbXxIBR+d1t&&8;nI8*h3zm{7t-v-1TE!B`?W}=c}dNL~(Zi{`r=Teg992
Gq5KmrAN&FU

diff --git a/salary_analytics/api.py b/salary_analytics/api.py
index 63c3560..8172a9b 100644
--- a/salary_analytics/api.py
+++ b/salary_analytics/api.py
@@ -6,15 +6,16 @@ from fastapi import FastAPI, HTTPException, BackgroundTasks, UploadFile, File
 from fastapi.responses import FileResponse
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
-from typing import Optional, Dict
+from typing import Optional, Dict, List
 import os
 import socket
 import logging
 import pandas as pd
 import tempfile
-
+from datetime import datetime
+from sqlalchemy import text
 from .main import SalaryAnalyticsPipeline
-from .config import OUTPUT_PATHS
+from .config import OUTPUT_PATHS, TABLE_NAME
 from .data_loader import DataLoader
 from .salary_predictor import SalaryPredictor
 from .salary_earner_analyzer import SalaryEarnerAnalyzer
@@ -56,6 +57,14 @@ class AnalysisResponse(BaseModel):
     data: Optional[Dict] = None
     file_path: Optional[str] = None
 
+class BatchResponse(BaseModel):
+    """Response model for batch processing."""
+    batch_number: int
+    total_batches: int
+    processed_rows: int
+    results_path: str
+    message: str
+
 def check_data_loaded():
     """Check if data is loaded before running analytics."""
     if pipeline.df is None:
@@ -277,4 +286,172 @@ async def load_data(source: str = "db", file: UploadFile = None):
         }
     except Exception as e:
         logger.error(f"Error loading data: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/run/streaming-pipeline", response_model=List[BatchResponse])
+async def run_streaming_pipeline(source: str = "db", file: UploadFile = None, batch_size: int = 10000):
+    """
+    Run the complete salary analytics pipeline in batches.
+    
+    Args:
+        source (str): Source of data ('db' or 'csv')
+        file (UploadFile): CSV file to load (required if source is 'csv')
+        batch_size (int): Number of rows to process in each batch
+    
+    Returns:
+        List[BatchResponse]: List of responses for each batch processed
+    """
+    try:
+        if source not in ['db', 'csv']:
+            raise HTTPException(status_code=400, detail="Source must be either 'db' or 'csv'")
+        
+        if source == 'csv' and not file:
+            raise HTTPException(status_code=400, detail="File must be provided when loading from CSV")
+        
+        # Initialize data loader
+        data_loader = DataLoader()
+        data_loader.chunk_size = batch_size
+        
+        # Create output directory for batch results
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        batch_output_dir = os.path.join(os.path.dirname(OUTPUT_PATHS['final_table']), f"batch_results_{timestamp}")
+        os.makedirs(batch_output_dir, exist_ok=True)
+        
+        responses = []
+        batch_number = 0
+        
+        def preprocess_chunk(chunk):
+            """Preprocess a chunk of data with the same logic as DataLoader."""
+            # Convert dates
+            chunk['trx_start_date'] = pd.to_datetime(chunk['trx_start_date'])
+            chunk['trx_end_date'] = pd.to_datetime(chunk['trx_end_date'])
+            
+            chunk = chunk.rename(columns={
+                    'd1': 'trx_type',
+                    'd2': 'trx_subtype',
+                    'd3': 'initiated_by',
+                    'd4': 'customer_id'
+                    })
+            chunk = chunk.dropna()
+            
+            return chunk
+        
+        if source == 'csv':
+            # Save uploaded file temporarily
+            with tempfile.NamedTemporaryFile(delete=False, suffix='.csv') as temp_file:
+                content = await file.read()
+                temp_file.write(content)
+                temp_file_path = temp_file.name
+            
+            try:
+                # Process CSV in chunks
+                for chunk in pd.read_csv(temp_file_path, chunksize=batch_size):
+                    batch_number += 1
+                    logger.info(f"Processing batch {batch_number}")
+                    
+                    # Preprocess chunk
+                    chunk = preprocess_chunk(chunk)
+                    
+                    # Run pipeline on chunk
+                    pipeline = SalaryAnalyticsPipeline()
+                    pipeline.df = chunk
+                    
+                    try:
+                        # Run analyses
+                        pipeline.run_keyword_analysis()
+                        pipeline.run_consistent_amount_analysis()
+                        pipeline.run_transaction_type_analysis()
+                        
+                        # Generate reports
+                        reports = pipeline.generate_salary_earner_reports()
+                        
+                        # Save batch results
+                        batch_results_path = os.path.join(batch_output_dir, f"batch_{batch_number}_results.csv")
+                        reports['final_table'].to_csv(batch_results_path, index=False)
+                        
+                        responses.append(BatchResponse(
+                            batch_number=batch_number,
+                            total_batches=-1,  # Unknown for CSV
+                            processed_rows=len(chunk),
+                            results_path=batch_results_path,
+                            message=f"Successfully processed batch {batch_number}"
+                        ))
+                    except Exception as e:
+                        logger.error(f"Error processing batch {batch_number}: {str(e)}")
+                        responses.append(BatchResponse(
+                            batch_number=batch_number,
+                            total_batches=-1,
+                            processed_rows=len(chunk),
+                            results_path="",
+                            message=f"Error processing batch {batch_number}: {str(e)}"
+                        ))
+            finally:
+                # Clean up temporary file
+                os.unlink(temp_file_path)
+        else:
+            # Process database in chunks
+            if not data_loader.connect():
+                raise HTTPException(status_code=500, detail="Failed to connect to database")
+            
+            # Get total row count
+            with data_loader.engine.connect() as conn:
+                count_query = text(f"SELECT COUNT(*) FROM {TABLE_NAME}")
+                total_rows = conn.execute(count_query).scalar()
+            
+            total_batches = (total_rows + batch_size - 1) // batch_size
+            offset = 0
+            
+            while offset < total_rows:
+                batch_number += 1
+                logger.info(f"Processing batch {batch_number} of {total_batches}")
+                
+                # Load chunk from database
+                query = f"SELECT * FROM {TABLE_NAME} LIMIT {batch_size} OFFSET {offset}"
+                chunk = pd.read_sql(query, data_loader.engine)
+                
+                if chunk.empty:
+                    break
+                
+                # Preprocess chunk
+                chunk = preprocess_chunk(chunk)
+                
+                # Run pipeline on chunk
+                pipeline = SalaryAnalyticsPipeline()
+                pipeline.df = chunk
+                
+                try:
+                    # Run analyses
+                    pipeline.run_keyword_analysis()
+                    pipeline.run_consistent_amount_analysis()
+                    pipeline.run_transaction_type_analysis()
+                    
+                    # Generate reports
+                    reports = pipeline.generate_salary_earner_reports()
+                    
+                    # Save batch results
+                    batch_results_path = os.path.join(batch_output_dir, f"batch_{batch_number}_results.csv")
+                    reports['final_table'].to_csv(batch_results_path, index=False)
+                    
+                    responses.append(BatchResponse(
+                        batch_number=batch_number,
+                        total_batches=total_batches,
+                        processed_rows=len(chunk),
+                        results_path=batch_results_path,
+                        message=f"Successfully processed batch {batch_number} of {total_batches}"
+                    ))
+                except Exception as e:
+                    logger.error(f"Error processing batch {batch_number}: {str(e)}")
+                    responses.append(BatchResponse(
+                        batch_number=batch_number,
+                        total_batches=total_batches,
+                        processed_rows=len(chunk),
+                        results_path="",
+                        message=f"Error processing batch {batch_number}: {str(e)}"
+                    ))
+                
+                offset += batch_size
+        
+        return responses
+    except Exception as e:
+        logger.error(f"Error in streaming pipeline: {str(e)}")
         raise HTTPException(status_code=500, detail=str(e)) 
\ No newline at end of file
diff --git a/salary_analytics/data_loader.py b/salary_analytics/data_loader.py
index 8eb6d25..ec2da46 100644
--- a/salary_analytics/data_loader.py
+++ b/salary_analytics/data_loader.py
@@ -69,6 +69,7 @@ class DataLoader:
                         'd4': 'customer_id'
                     })
                 
+                chunk = chunk.dropna()
                 chunks.append(chunk)
             
             # Combine all chunks
@@ -127,6 +128,7 @@ class DataLoader:
                     'd4': 'customer_id'
                 })
                 
+                chunk = chunk.dropna()
                 chunks.append(chunk)
                 offset += self.chunk_size
                 
diff --git a/salary_analytics/salary_earner_analyzer.py b/salary_analytics/salary_earner_analyzer.py
index 00407ba..ccdc6bb 100644
--- a/salary_analytics/salary_earner_analyzer.py
+++ b/salary_analytics/salary_earner_analyzer.py
@@ -18,7 +18,7 @@ class SalaryEarnerAnalyzer:
     def filter_venn_section(self, **kwargs):
         """Filter accounts based on specified combinations of hypothesis flags."""
         valid_columns = {'is_salary_related', 'is_consistent_amount', 'is_salary_type'}
-        df1 = self.df[self.df['initiated_by'] == 'C']
+        df1 = self.df[self.df['initiated_by'] == 'C'].copy()
         
         invalid_keys = set(kwargs.keys()) - valid_columns
         if invalid_keys:
@@ -28,7 +28,13 @@ class SalaryEarnerAnalyzer:
         for key, value in kwargs.items():
             condition &= (df1[key] == value)
 
-        return df1[condition]
+        filtered_df = df1[condition]
+        
+        # Drop any rows with NaN values in critical columns
+        critical_cols = ['accountid', 'trx_start_date', 'amount']
+        filtered_df = filtered_df.dropna(subset=critical_cols)
+        
+        return filtered_df
 
     def plot_hypothesis_overlap(self, hypothesis1_df, hypothesis3_df, hypothesis4_df, account_col='accountid'):
         """Plot and save Venn diagram showing overlap between hypotheses."""
@@ -47,21 +53,37 @@ class SalaryEarnerAnalyzer:
         """Generate a table of salary earners with their metrics."""
         results = []
         for accountid, group in all_three_hypotheses.groupby('accountid'):
+            # Skip if group is empty
+            if group.empty:
+                continue
+                
             # Calculate required metrics
             num_months = len(group)
+            
+            # Handle last 6 months calculation
             last_6_months = group[group['trx_start_date'] >= (datetime.now() - timedelta(days=180))]
-            least_inflow = last_6_months['amount'].min()
-            avg_salary = group['amount'].mean()
-
-            # Calculate days since last transaction
+            if last_6_months.empty:
+                least_inflow = 0
+            else:
+                least_inflow = last_6_months['amount'].min()
+            
+            # Handle average salary calculation
+            if group['amount'].notna().any():
+                avg_salary = group['amount'].mean()
+            else:
+                avg_salary = 0
+            
+            # Calculate days_since_last_trx with NaN handling
             group['days_since_last_trx'] = group['trx_start_date'].diff().dt.days
             median_interval = group['days_since_last_trx'].median()
+            if pd.isna(median_interval):
+                median_interval = 30  # Default to 30 days if no interval data
 
             last_date = group['trx_start_date'].max()
             next_date = last_date + timedelta(days=median_interval)
             next_amount = avg_salary
 
-            # Boolean flags
+            # Boolean flags with NaN handling
             days_since_last = (datetime.now() - last_date).days
             has_45d = days_since_last <= 45
             has_2m = len(group[group['trx_start_date'] >= (datetime.now() - timedelta(days=60))]) >= 2
@@ -78,7 +100,9 @@ class SalaryEarnerAnalyzer:
             })
 
         final_df = pd.DataFrame(results)
-        final_df = final_df.dropna()
+        # Drop rows where all numeric columns are NaN
+        numeric_cols = ['num_months', 'least_inflow_6m', 'avg_monthly_salary', 'estimated_next_amount']
+        final_df = final_df.dropna(subset=numeric_cols, how='all')
         return final_df
 
     def analyze_salary_earners(self, final_df):