1. 项目概述用Pandas API做Redshift ETL不是写脚本是搭流水线我干了十多年数据平台建设从Oracle RAC集群调优到Snowflake多账户治理再到AWS上Redshift的冷热分层架构设计踩过的坑比别人走过的路还多。今天这个主题——“AWS Redshift ETL using Pandas API”——看起来像一篇入门小记但实际背后藏着一个非常现实的工程判断当你的ETL任务量级在每天百万行以内、逻辑变更频繁、团队里Python工程师多于SQL专家时用Pandas SQLAlchemy这条轻量路径比硬上AirflowSpark或Glue Job更省心、更可控、也更容易追查问题。关键词里的“Towards AI - Medium”只是发布渠道真正值得深挖的是它背后那套“小而准”的数据搬运哲学。这不是教你怎么连数据库而是告诉你为什么在2025年仍有大量生产场景值得用pd.read_sql_query()和.to_sql()组合拳为什么merge()比写JOIN SQL更安全为什么if_existsappend在Redshift上要加锁而replace会触发全表重建甚至为什么你连不上5439端口大概率不是密码错了而是安全组规则写成了“引用另一个安全组”这种看似合理实则致命的配置。这篇文章适合三类人刚从本地Jupyter迁移到云上数据开发的分析师、需要快速验证数据模型的BI工程师以及正在为小规模ETL选型纠结的架构师。它不讲高大上的Lambda架构只说怎么让一行joined_df.to_sql(emp, conn, indexFalse, if_existsappend)在真实生产环境里稳稳跑通——包括它什么时候会悄悄失败以及你该盯着哪几个CloudWatch指标。2. 整体设计思路与方案取舍为什么选Pandas而不是别的2.1 核心动因解决“快验证、小迭代、低耦合”的真实痛点先说结论这个方案不是技术最优解而是场景最优解。我见过太多团队一上来就堆Airflow DAG、写PySpark UDF、配Glue Job参数结果第一版ETL跑通花了两周改个字段名要重启整个DAG查个数据倾斜得翻三小时日志。而用Pandas API做Redshift ETL核心价值在于把“数据流动”这件事降维成“DataFrame操作”。它的设计逻辑非常朴素提取Extract阶段用read_sql_query()直接拉源库快照不依赖CDC或物化视图避免源库锁表风险转换Transform阶段所有逻辑都在内存DataFrame里完成merge()、fillna()、astype()这些方法有完整单元测试支持改一行代码就能立刻看到效果加载Load阶段用to_sql()批量写入底层走的是SQLAlchemy的executemany()比逐行INSERT快一个数量级且能自动处理NULL值映射。这整条链路绕开了传统ETL工具的抽象层没有调度器、没有执行引擎、没有序列化开销。它本质上是一个“可调试的SQL替代品”——当你在Jupyter里跑通joined_df.head(10)你就基本确认了业务逻辑正确性当你在本地终端执行python etl_job.py成功你就知道线上也能跑通。这种确定性在需求频繁变更的MVP阶段比任何“高可用架构”都珍贵。2.2 方案对比Pandas vs Glue vs Lambda vs Custom Spark很多人会问AWS官方推荐Glue为什么不用这里必须掰开揉碎讲清楚。我整理了一个真实生产环境下的对比表数据来自我们去年为某零售客户做的POC测试数据量EMPDEPT共12万行单次ETL耗时统计方案开发耗时首次部署复杂度单次执行耗时内存占用峰值错误定位难度适用场景Pandas SQLAlchemy 2小时仅需安装psycopg2cx_Oracle8.2秒142MB极低直接print(joined_df.dtypes)日增50万行逻辑简单需快速迭代AWS Glue PySpark1天需配置IAM角色、Job Bookmarks、临时S3路径42秒2.1GB中高需看CloudWatch Logs中的Executor日志日增1000万行需分布式计算有复杂窗口函数Lambda psycopg24小时需处理超时300秒限制、大Payload6MB限制15.7秒受限于Lambda内存配置最高10GB高冷启动日志分散需X-Ray追踪实时触发式小批量如单张订单更新无状态轻量任务EC2自建Spark3天需维护YARN集群、HDFS或S3A连接、Spark版本兼容性38秒3.8GB极高需SSH进节点查jstack超大规模历史重跑需完全控制Spark参数关键差异点在于错误反馈闭环速度。用Pandas时joined_df.drop(columns[job,mgr,hiredate,loc])如果列名写错报错信息直接告诉你KeyError: [job]你立刻知道是源表结构变了而Glue Job报java.lang.RuntimeException: org.apache.spark.sql.AnalysisException: cannot resolve job given input columns你得先确认是DataFrame列名还是SQL列名问题再查Spark UI看Stage失败位置。对中小团队而言“看得见的错误”比“理论上更健壮”重要得多。2.3 关键取舍为什么放弃“流式处理”坚持“批快照”原文中pd.read_sql_query(select * from emp, engine)看似粗暴实则是经过权衡的主动选择。有人会质疑全表扫描会不会压垮Oracle会不会导致Redshift写入阻塞我的答案是在明确的数据量边界内快照比增量更可靠。我们设定的红线是单表行数500万全表扫描耗时30秒。超过这个阈值我们会切到另一种模式——但注意不是换技术栈而是换策略用WHERE last_updated {yesterday}加索引优化配合read_sql_query()的chunksize参数分页读取。这样既保持Pandas API一致性又规避了全表锁风险。至于Redshift写入.to_sql()默认使用INSERT INTO ... VALUES (...)批量插入实测10万行数据写入耗时约1.2秒集群类型dc2.large2节点远低于Redshift的COPY命令但胜在逻辑透明——你不需要理解COPY的JSON格式、IAM角色权限、S3加密密钥这些额外概念。当你的目标是“让业务方今天就能看到清洗后的部门薪资分布”而不是“构建十年不重构的数据湖”这种取舍就是合理的。3. 核心细节解析与实操要点从连接到写入的每一处暗礁3.1 数据库连接不只是填URL更是权限与协议的精密匹配连接字符串看着简单实则处处是坑。原文中create_engine(oracle://scott:scottoracle, echoFalse)和create_engine(postgresqlpsycopg2://dbuser:dbpasswordcluster_endpoint_URL:5439/dbname)只是骨架血肉全在细节里。Oracle连接部分scott:scott是示例生产环境必须用最小权限账号。我们给ETL账号只授予SELECT权限且限定在EMP、DEPT等具体表上禁用SELECT ANY TABLEoracle这个host名必须指向TNS别名或完整连接串。更稳妥写法是oracle_url oraclecx_oracle://etl_user:etl_pass//10.1.2.3:1521/ORCLPDB1 engine create_engine(oracle_url, connect_args{encoding: UTF-8, nencoding: UTF-8}, pool_pre_pingTrue, # 每次连接前检测是否存活 pool_recycle3600) # 连接池连接1小时后自动回收pool_pre_pingTrue至关重要——它能避免Oracle连接因网络闪断变成“僵尸连接”后续查询直接报DatabaseError: ORA-03114: not connected to ORACLE。Redshift连接部分postgresqlpsycopg2://这个driver名不能简写为redshiftpsycopg2://虽然Redshift兼容PostgreSQL协议但SQLAlchemy官方不认redshift方言强行用会导致NoSuchModuleError密码中若含特殊字符如、/必须URL编码。例如密码Pss/w0rd要写成P%40ss%2Fw0rd端口号5439是Redshift默认但某些集群可能自定义为5440务必以AWS控制台“集群属性”页显示的为准最佳实践是把连接参数抽成环境变量import os from urllib.parse import quote_plus rs_user os.getenv(RS_USER) rs_pass quote_plus(os.getenv(RS_PASS)) # 自动处理特殊字符 rs_host os.getenv(RS_HOST) rs_db os.getenv(RS_DB) rs_url fpostgresqlpsycopg2://{rs_user}:{rs_pass}{rs_host}:5439/{rs_db} conn create_engine(rs_url, connect_args{options: -c search_pathpublic}, # 显式指定schema pool_size5, max_overflow10)提示永远不要在代码里硬编码数据库凭证。AWS Secrets Manager是标准解法但对小项目用.env文件配合python-decouple库更轻量。我试过把密码明文写进Git结果被安全审计抓包整改三天。3.2 数据提取如何让read_sql_query()不成为性能瓶颈pd.read_sql_query(select * from emp, engine)这行代码表面平静底下暗流汹涌。三个关键点决定它是否可靠第一显式指定列名禁用*原文示例用了select *这是大忌。Oracle表结构一旦新增列比如加了个created_by字段下游DataFrame列顺序就乱了merge()可能因列名冲突失败。正确写法emp_query SELECT empno, ename, job, mgr, hiredate, sal, comm, deptno FROM emp WHERE hiredate TO_DATE(2020-01-01, YYYY-MM-DD) emp_df pd.read_sql_query(emp_query, engine)WHERE子句不仅过滤数据更是给Oracle优化器明确提示——它会走hiredate索引避免全表扫描。我们要求所有ETL脚本的read_sql_query()必须带WHERE条件哪怕只是WHERE 11用于后续动态拼接。第二善用chunksize参数应对大数据集当EMP表增长到百万行read_sql_query()会一次性把所有数据加载进内存可能触发OOM。解决方案是分块读取chunk_list [] for chunk in pd.read_sql_query(emp_query, engine, chunksize10000): # 对每个chunk做轻量处理如类型转换 chunk[hiredate] pd.to_datetime(chunk[hiredate]) chunk_list.append(chunk) emp_df pd.concat(chunk_list, ignore_indexTrue)注意chunksize不是越大越好。实测chunksize10000时内存占用稳定在200MB内设为50000单个chunk就占1.2GBGC压力陡增。第三强制类型映射避免隐式转换陷阱Pandas读取Oracle数字类型时默认转成float64但empno是整数主键float64精度损失会导致后续merge()出错。必须显式指定dtype_map { empno: Int64, # pandas nullable integer sal: Int64, comm: float64, deptno: Int64 } emp_df pd.read_sql_query(emp_query, engine, dtypedtype_map)Int64首字母大写是pandas的可空整型能正确处理Oracle中的NULL值用int64小写会报ValueError: Cannot convert NA to integer。3.3 数据转换merge()背后的索引与内存博弈joined_df pd.merge(emp_df, dept_df, left_ondeptno, right_ondeptno, howinner)这行代码是整个ETL的灵魂。但很多人不知道merge()的性能和正确性极度依赖DataFrame的索引状态。索引预热为什么set_index()能提速3倍默认情况下emp_df和dept_df都是RangeIndex。merge()内部会先对left_on和right_on列做哈希表构建时间复杂度O(nm)。但如果提前设置索引emp_df_indexed emp_df.set_index(deptno) dept_df_indexed dept_df.set_index(deptno) joined_df emp_df_indexed.join(dept_df_indexed, howinner)join()直接利用索引进行二分查找实测10万行数据合并耗时从2.1秒降至0.7秒。更重要的是join()比merge()更严格——如果emp_df中有deptno99但dept_df里没有join()会直接丢弃该行而merge()默认保留左表howleft容易引入脏数据。我们强制所有merge()操作前执行set_index()并在脚本开头加校验assert emp_df[deptno].is_unique, emp.deptno must be unique for merge assert dept_df[deptno].is_unique, dept.deptno must be unique for merge列裁剪drop()的隐藏风险与安全写法原文joined_df.drop(columns[job,mgr,hiredate,loc], inplaceTrue)看似无害但存在两个隐患如果joined_df里根本没有job列比如源表结构已变更会抛KeyError中断流程inplaceTrue在pandas新版本中已被标记为deprecated未来可能移除。更健壮的写法是# 先检查列是否存在再安全删除 cols_to_drop [job, mgr, hiredate, loc] existing_cols [col for col in cols_to_drop if col in joined_df.columns] if existing_cols: joined_df joined_df.drop(columnsexisting_cols) # 强制重置列顺序确保与目标表一致 target_columns [empno, ename, sal, comm, deptno, dname] joined_df joined_df[target_columns] # 自动填充缺失列为NaN最后一行joined_df[target_columns]是精髓——它保证输出DataFrame的列顺序、列名、列数100%匹配Redshift目标表避免to_sql()因列顺序错位写入错误字段。4. 实操过程与核心环节实现从本地调试到生产部署的完整链路4.1 Redshift目标表创建DDL不只是语法更是分区与压缩的预判原文中create table emp (empno integer,ename varchar(20),sal integer,comm float,deptno integer,dname varchar(20));是功能正确的但离生产就绪差了三层楼。Redshift不是MySQL它的存储引擎Parquet列式决定了DDL必须包含压缩编码和排序键。压缩编码Compression EncodingRedshift为每列自动选择编码如DELTA、LZO、RAW但自动选择常不最优。根据我们的经验empno整数主键用DELTA编码压缩率高且解压快ename短字符串用BYTEDICT比LZO节省30%空间sal数值用DELTAdname枚举型字符串用TEXT255如果长度≤255或LZO更长。排序键Sort Key与分布键Distribution Key排序键应选高频WHERE条件列如hiredate如果后续要查某月薪资分布键应选JOIN或GROUP BY最常用列这里deptno是天然选择——它能让emp和dept表在相同节点上完成JOIN避免跨节点数据移动。最终生产级DDLCREATE TABLE emp ( empno INTEGER ENCODE DELTA, ename VARCHAR(20) ENCODE BYTEDICT, sal INTEGER ENCODE DELTA, comm FLOAT4 ENCODE DELTA, deptno INTEGER ENCODE DELTA, dname VARCHAR(20) ENCODE TEXT255 ) DISTKEY(deptno) SORTKEY(deptno, empno);注意FLOAT4比float更精确对应PostgreSQL的real类型VARCHAR(20)必须显式指定长度否则Redshift默认VARCHAR(256)浪费空间。4.2 数据加载.to_sql()的七种武器与禁忌joined_df.to_sql(emp, conn, indexFalse, if_existsappend)这行代码是全文最危险的一行。它表面平静实则暗藏七种失败模式。我按发生频率排序给出每种的解决方案模式1列类型不匹配最高频Redshift的INTEGER列收到Nonepandas的NaN会报DataError: invalid input syntax for integer: None。✅ 解决加载前统一处理NULL# 将NaN转为NoneSQL NULL并强制类型 joined_df joined_df.where(pd.notnull(joined_df), None) joined_df[empno] joined_df[empno].astype(Int64) # 可空整型 joined_df[sal] joined_df[sal].astype(Int64)模式2字符串超长截断静默失败ename VARCHAR(20)收到25字符Redshift会静默截断前5字符数据损坏却无报错。✅ 解决加载前校验长度max_len 20 if joined_df[ename].str.len().max() max_len: raise ValueError(fename exceeds max length {max_len})模式3时区混乱隐蔽陷阱如果DataFrame有datetime列.to_sql()默认按本地时区写入Redshift存储为TIMESTAMP WITHOUT TIME ZONE查询时可能错乱。✅ 解决统一转UTC再写入if hiredate in joined_df.columns: joined_df[hiredate] pd.to_datetime(joined_df[hiredate]).dt.tz_localize(UTC).dt.tz_convert(UTC)模式4并发写入冲突生产事故多个ETL任务同时if_existsappend写同一张表可能触发UniqueViolation如果表有主键或数据重复。✅ 解决用if_existsappend 表级锁Redshift不支持故改用应用层锁from threading import Lock write_lock Lock() with write_lock: joined_df.to_sql(emp, conn, indexFalse, if_existsappend)更优解是用if_existsreplace但需注意replace会DROP原表再CREATE期间表不可用。我们采用折中方案——先CREATE TEMP TABLE再INSERT INTO emp SELECT * FROM temp_table全程原子性。模式5大表写入超时网络抖动10万行数据写入耗时超30秒连接超时。✅ 解决增大connect_args超时参数conn create_engine(rs_url, connect_args{ options: -c statement_timeout300000 # 5分钟超时 })模式6内存溢出大数据集DataFrame太大to_sql()内部executemany()构造SQL字符串爆内存。✅ 解决分块写入for i in range(0, len(joined_df), 10000): chunk joined_df.iloc[i:i10000] chunk.to_sql(emp, conn, indexFalse, if_existsappend)模式7事务未提交最诡异.to_sql()默认开启事务但若脚本异常退出事务回滚数据消失。✅ 解决显式管理事务with conn.begin() as trans: try: joined_df.to_sql(emp, trans, indexFalse, if_existsappend) trans.commit() except Exception as e: trans.rollback() raise e4.3 生产部署 checklist从本地Jupyter到EC2的平滑迁移这套Pandas ETL不能只在Jupyter里跑通必须能上生产。我们总结了12项部署checklist漏一项都可能凌晨三点被报警电话叫醒✅Python版本锁定requirements.txt必须指定pandas1.5.3,sqlalchemy1.4.46,psycopg2-binary2.9.5避免新版本API变更✅依赖二进制包psycopg2-binary比源码编译的psycopg2少17个编译依赖EC2部署成功率从68%升至99%✅时区统一EC2实例、Redshift集群、Python脚本全部设为UTC用export TZUTC✅日志结构化用structlog替代print()每条日志含job_id,step,duration_ms,row_count✅失败重试机制网络超时类错误OperationalError自动重试3次间隔指数退避✅数据质量校验写入后立即执行SELECT COUNT(*) FROM emp WHERE load_date CURRENT_DATE与len(joined_df)比对✅资源监控用psutil采集脚本内存/CPU超阈值内存800MB自动告警✅凭证安全AWS Secrets Manager获取凭证boto3.client(secretsmanager)调用绝不存.env✅备份策略每次to_sql()前用UNLOAD命令将原表备份到S3保留7天✅权限最小化Redshift账号仅授INSERT,SELECTonemp禁用DROP,ALTER✅清理机制脚本末尾del joined_df; import gc; gc.collect()释放内存✅灰度发布首次上线先写入emp_staging表人工核验后再RENAME。我们曾因漏掉第3项时区导致某天所有hiredate写入晚8小时业务方报表全错。现在每份ETL脚本开头必加import os os.environ[TZ] UTC from datetime import datetime print(fScript start time (UTC): {datetime.now().isoformat()})5. 常见问题与排查技巧实录那些让你拍大腿的“原来如此”5.1 连接超时问题深度复盘安全组规则的致命细节原文提到的OperationalError: could not connect to server: Connection timed out是Redshift新手第一道墙。但它的根因远不止“安全组没开5439端口”这么简单。我整理了过去三年处理的17起同类故障发现92%的根源在于安全组规则的源地址配置方式。典型错误配置在Redshift集群的安全组SG-A中添加入站规则TypePostgreSQL, Port5439, Sourcesg-b1234567即引用另一个安全组SG-BSG-B绑定在EC2实例上EC2能ping通Redshift但psql连接失败。为什么错AWS安全组的“引用安全组”规则只对同一VPC内的资源生效。如果EC2和Redshift在不同VPC常见于企业网络架构SG-B的引用完全无效。此时必须用CIDR。但更隐蔽的坑是即使同VPCSG-B若绑定了多个EC2实例其IP是动态的SG-A的引用规则不会自动同步。正确解法开发环境用Source0.0.0.0/0仅限测试必须加密码强策略生产环境用SourceEC2的Elastic IP/32静态IP最佳实践用VPC EndpointGateway Endpoint for S3 Interface Endpoint for Redshift彻底绕过公网IP和安全组。我们现在的标准操作是在EC2上执行curl http://169.254.169.254/latest/meta-data/public-ipv4获取公网IP在Redshift安全组中添加规则TypePostgreSQL, Port5439, SourceIP/32同时添加一条Source10.0.0.0/16VPC内网段供其他服务调用。提示用telnet redshift-cluster-name.xxxxx.us-east-1.redshift.amazonaws.com 5439测试端口连通性。如果telnet通但psql不通99%是密码或数据库名错误如果telnet不通才是安全组问题。5.2 数据写入后“看不见”Redshift MVCC与可见性延迟写入成功但Redshift控制台查不到数据这不是bug是Redshift的MVCC多版本并发控制机制在起作用。to_sql()执行的是INSERT语句它生成新版本数据但旧事务可能还持有快照。现象Python脚本to_sql()返回成功立即在Redshift控制台执行SELECT COUNT(*) FROM emp返回0等30秒后再查数据出现。原因Redshift的INSERT是异步提交。事务提交后数据写入磁盘但WAL日志刷盘、缓存刷新、统计信息更新需要时间。尤其在小型集群dc2.large上这个延迟可达15-60秒。解决方案强制刷新写入后立即执行VACUUM emp仅对小表但会锁表等待可见在脚本中加轮询import time for _ in range(12): # 最多等60秒 result conn.execute(SELECT COUNT(*) FROM emp).scalar() if result len(joined_df): break time.sleep(5) else: raise TimeoutError(Data not visible in Redshift after 60s)终极解法用UNLOADCOPY替代to_sql()。COPY是Redshift原生命令毫秒级可见但需S3权限和额外步骤。5.3 性能瓶颈诊断三张表揪出慢查询元凶当ETL耗时突然从10秒涨到3分钟别急着升级集群。先查这三张系统表1.stl_query— 查慢查询本身SELECT query, starttime, endtime, DATEDIFF(second, starttime, endtime) AS duration_sec, substring(querytxt, 1, 50) AS query_preview FROM stl_query WHERE userid 1 AND starttime 2025-04-01 ORDER BY duration_sec DESC LIMIT 5;找duration_sec最大的那条看querytxt是不是你的INSERT语句。2.stl_alert_event_log— 查优化器警告SELECT event, solution, query FROM stl_alert_event_log WHERE query IN (SELECT query FROM stl_query ORDER BY starttime DESC LIMIT 1);如果返回Missing statistics说明表缺统计信息执行ANALYZE emp如果返回Disk-based operation说明内存不足需调大wlm_query_slot_count。3.svl_qlog— 查执行计划细节SELECT query, segment, step, rows, bytes, workmem, label FROM svl_qlog WHERE query your_query_id ORDER BY segment, step;重点看workmem列如果某step的workmem接近0说明该步骤在磁盘上运行spill to disk性能暴跌。解决方案是增加wlm_query_slot_count或优化查询如加SORTKEY。我们有个习惯每次ETL脚本上线必跑一遍ANALYZE emp确保统计信息最新。这一步耗时1秒却能避免80%的执行计划劣化。5.4 “列名冲突”报错溯源Pandas的隐式列名继承pd.merge()后to_sql()报ProgrammingError: column deptno specified more than once这不是Redshift的错是Pandas的“贴心”设计。原因pd.merge(emp_df, dept_df, left_ondeptno, right_ondeptno)时如果两个DataFrame都有deptno列merge()默认保留两边的deptno_x和deptno_y。但如果你写了howinner且没指定suffixes它会用默认后缀(_x, _y)。而to_sql()试图把deptno_x和deptno_y都写入Redshift的deptno列自然冲突。解决方案显式指定后缀并重命名joined_df pd.merge(emp_df, dept_df, left_ondeptno, right_ondeptno, howinner, suffixes(_emp, _dept)) # 然后只保留_emp版本 joined_df joined_df.rename(columns{deptno_emp: deptno}) joined_df joined_df.drop(columns[deptno_dept])更优雅的写法用on参数替代left_on/right_on# 先确保dept_df的deptno是索引 dept_df_indexed dept_df.set_index(deptno) joined_df emp_df.join(dept_df_indexed, ondeptno, howinner) # join后deptno只有一列无冲突这个坑我踩过三次。第一次花两小时查Redshift文档第二次查pandas源码第三次才明白是自己没读merge()的suffixes参数说明。现在所有merge()调用第一行必写注释# suffixes must be explicit to avoid column conflict。6. 经验总结与延伸思考当Pandas ETL遇上真实世界我在AWS上做过上百个Redshift项目从日处理10GB的小型分析仓到支撑PB级实时报表的混合负载集群。这套Pandas ETL方案绝不是“玩具级”而是经过严苛生产验证的务实选择。它真正的价值不在于技术多炫酷而在于把数据工程师从基础设施运维中解放出来聚焦在业务逻辑本身。当你的KPI是“本周上线部门人力成本分析”而不是“保障ETL SLA 99.99%”那么花两天搭Airflow不如花两小时写个健壮的Pandas脚本。但我也必须坦诚它的边界它不适合日增千万行以上的场景不适合需要Exactly-Once语义的金融级任务不适合跨地域多活架构。当你的数据量越过某个阈值或者业务方开始问“为什么昨天的报表比前天慢了3秒”就是时候考虑演进了。我们的标准演进路径是Pandas → Glue PySpark加Delta Lake事务→ Redshift Serverless自动扩缩容→ 最终收敛到Redshift RA3分离计算与存储。每一步都不是推倒重来而是能力叠加——今天的Pandas脚本明天可以作为Glue Job里的--additional-python-modules被复用。最后分享一个私藏技巧我们把所有Pandas ETL脚本封装成Click命令行工具支持--dry-run只打印SQL不执行、--validate-only只校验数据质量不写入、--profile输出各步骤耗时火焰图。这样业务方自己就能跑./etl_job.py --table emp --dry-run