企业官网建设流程全解析

我在处理一个每天要往HBase里写几百万行数据的批任务时，第一次意识到手写的shell调度已经撑不住了。任务之间要串行等待、要重试、要判断上游数据是否就绪，最后还要做行数校验。当时正好在用Python写ETL，就顺手把工作流切到了Luigi。Luigi是Python生态里一个很实用的工作流引擎，核心思路是“每个任务都有明确依赖，每个产出都有明确标记”，配上HBase做存储层，一套批处理链路能跑得比较省心。

听起来像“调度器加任务库”，但实际用下来，Luigi更像是一套把任务依赖、失败处理、幂等重跑全部代码化的框架。它不负责实时计算，也不内置复杂的可视化大屏，但胜在足够轻、足够贴近数据开发的直觉：你定义Task，声明这个Task需要哪些上游，在Run里写真正干活的代码，然后交给调度器编排。

这篇文章不是从零讲Python，而是从一个真实场景出发：用HBase存储指标数据，用Luigi编排“生成数据 → 写入HBase → 扫描校验 → 产出结果文件”的完整流程。适合谁看？正在用Python写批处理、对HBase有一定了解、想摆脱cron硬编码依赖的人。就算你已经在用其他调度框架，也可以看看Luigi这套设计里有哪些思路值得借鉴。好，从选型开始讲。

1. 为什么是Luigi，而不是XX？——工作流选型背后的逻辑

1.1 cron脚本和多步任务的痛点

很多团队一开始都是从cron起家的。一个shell脚本，凌晨两点跑，先调接口拉数据，再处理，然后写HBase，最后发通知。前几个月都没问题，直到某天HBase节点抖动，脚本在写一半的时候挂了。第二天一看，数据少了几个region的量，没有任何标记，你甚至不知道任务到底有没有跑完。

cron最大的问题不是“定时”，而是没有“状态”。它不知道上一步成功还是失败，不知道这一步的输出能不能被下一步消费，更不知道任务重跑时会不会重复写入。你会发现要自己维护日志文件、锁文件、状态文件，这些临时方案越滚越复杂，到最后所有人在一堆命名奇怪的文件里猜今天的任务到底跑到哪了。

Luigi把这个问题抽象得很简单：每个Task有一个或者多个output，只有output完整存在才认为这个Task完成；每个Task可以声明requires，上游没完成，下游就不会启动。这样一来，整套批处理的状态管理就从“文件系统里靠猜”变成了“代码里可查”。

1.2 Luigi的核心设计：Task、Target、Parameter

刚接触Luigi的人可能会被这三个词绕晕，其实对应关系很直接：

• Task：一个可执行的工作单元。继承luigi.Task，实现requires()返回依赖，实现output()返回目标，实现run()写真正的逻辑。
• Target：任务的产出物标记。不一定是文件，可以是HDFS文件、数据库记录、对象存储对象，也可以是HBase里的一个状态行。只要这个Target存在，任务就认为“我做过了”。
• Parameter：任务的入参。比如日期参数、批次号。同一个Task类传不同参数就是不同任务实例，Luigi在依赖图里会严格区分它们。

这个设计的精髓在于：任务之间通过Target建立依赖，而不是通过“内存里互相调函数”。今天跑一次，明天重跑一次，只要Target判定不可靠，调度器就会帮你重做。HBase这种外部存储恰好很适合和Target搭配，后面我会专门展开讲。

1.3 为什么不选Airflow

现在Airflow名气很大，很多人一上来就问“直接上Airflow不好吗”。我的看法是，看场景。

Airflow全家桶包含Web UI、DAG可视化、插件体系、多执行器，这些功能确实强，但也带来部署成本和高运行依赖。如果只是几十个批任务、团队又主要写Python和SQL，Luigi的启动成本明显更低：安装依赖、写一个task文件、起一个luigid进程，就完了。

再一个区别是执行模型。Airflow以DAG为中心，调度器全局统筹；Luigi以Task为中心，每个Task自己声明依赖，调度器按依赖图逐个执行。做大规模分布式调度时Airflow更适合，但如果任务边界清晰、以Python函数为主，Luigi写出来的代码量更少，调试也更直接。

提示：选型没有绝对好坏。如果团队未来要做复杂的DAG管理、多团队共用、严格权限控制，建议直接上Airflow；如果任务是“重数据、轻编排”，Luigi这套“代码即工作流”的风格反而更好维护。我见过太多团队为了一个页面的可视化，引入一大套调度系统，最后天天在处理调度系统本身的问题。

2. 先搞定HBase：集成前的基础准备

2.1 HBase的接入方式：Java API和Thrift/REST

HBase写数据最稳的方案是Java API，能用到协处理器、完整过滤器堆栈，吞吐也最高。但我们的场景是Python调度和Python ETL，不可能为了几个批任务专门写一层Java服务。实际项目里，大部分Python接入HBase的路径是走Thrift网关或REST网关。

Thrift网关是HBase提供的一个独立服务，把客户端的Put、Get、Scan转成Java API调用。它的优点是与Java API的语义几乎一致，Filter、Batch、RowKey范围都能支持；缺点是Thrift版本和HBase版本必须匹配，踩过坑的人都知道，版本对不上会出现各种诡异的字段错位。REST网关是另一个可选方案，但性能一般，适合低并发查询。批处理写几百万行这种负载，建议直接走Thrift。

如果你所在环境里，HBase是托管服务，运维层面往往已经开了Thrift端口，你需要确认的是端口号、协议类型和是否有鉴权。如果是自己搭的集群，要记得单独启动Thrift服务进程，这个服务不会跟着RegionServer自动开。

2.2 Python侧选型：happybase

Python生态里接HBase Thrift，现在最常用的就是happybase类库。它对Thrift协议做了封装，API设计得很像数据库连接。连接、建表、put、scan都有对应的pythonic写法，批量写用的batch提交和HBase的batch语义一致，能把多次RPC合成一次，吞吐提升非常明显。

另一个选择是自己拿Thrift生成客户端，或者用hbase-thrift这类库。不是不行，但要自己处理连接池、异常重试，性价比不高。除非公司内部已经有封装好的公共包，否则从零接入我建议直接用happybase起步。

选库之外，还要注意HBase服务端是否启用了Thrift，以及Thrift版本。如果运维是独立部署的，问清楚Thrift端口、协议类型（二进制还是framed），这些信息在后面连不上时会救命。

2.3 连接参数、连接池与超时

happybase.Connection的第一个参数是Thrift主机，后面跟着port、timeout、table_names等参数。timeout默认是10000毫秒，注意这个超时只是socket读超时，不是整批任务的总超时。批量写入时，一次batch可能包含几千上万行，在集群压力大时单次RPC会超过10秒，这时候直接把timeout调大不一定解决问题，更好的做法是缩小batch的size。

连接池方面，happybase没有内置复杂的池化，常规做法是一个Task的run里建立连接、用完关闭。如果遇到需要反复拿连接的任务，可以用简单的连接池封装，或者利用happybase.Connection的延迟连接特性，在run里只建一次。不要在每个循环里都new一个连接，那是把时间和资源都浪费在握手上了。

注意：HBase Thrift连接在长时间空闲后容易被服务端断开。写批任务时，如果Scan要跑很久，务必在循环里捕获异常并重建连接，否则会出现“任务跑了一半，连接死了”的情况。后面排查章节我还会展开讲。

3. 设计一个HBase写入工作流：从任务拆分到代码落地

3.1 场景定义：每日指标数据入HBase

先说一个我在模拟项目X里常用的场景，它足够典型，又能覆盖Luigi和HBase集成的几个关键点。某业务每天凌晨会产生一批订单指标，格式是一行一个指标：日期、商户ID、指标名、数值。数据源是一个外部文本文件，处理之后要写入HBase。表结构用“日期 + 商户ID”做行键前缀，列族叫cf，列名是指标名。

任务的目标很明确：每天都要生成当天的HBase数据。可能前一天处理到一半挂了，也可能某天数据源更新了需要重跑，还可能跑完以后要校验行数和关键指标的汇总值。这些诉求刚好是Luigi的强项。

3.2 Task拆分：CheckSource、GenerateRows、WriteHBase、VerifyCount

这种场景我不会只用一个巨无霸Task把所有代码塞进去，而是拆成四个Task，让依赖关系自动形成管线：

• CheckSource：检查当天源文件是否已就绪。文件没出现，任务直接失败，不会往下走。
• GenerateRows：把源文件解析成标准行，输出一份中间文件。这一步是纯函数式的，可以本地跑，方便单测。
• WriteHBase：读取中间文件，批量写入HBase，完成后打标记。
• VerifyCount：扫描HBase中当天行键前缀的行数，和GenerateRows的行数做比对，不一致就抛异常。

为什么要多拆一个GenerateRows层？因为写入HBase之前，我们需要一个“可检查的中间产物”。如果直接把源文件读到内存再写，一旦写了一半挂了，没有任何现场可以检查。有了中间文件，问题在于任务失败时，看这个清理出的数据文件就能定位问题。

3.3 关键代码：Luigi Task定义

先定义一个基类，把日期参数统一放进去：

import luigi class DailyTask(luigi.Task): date = luigi.DateParameter(default=luigi.date.today())

接着是CheckSource和GenerateRows：

import os import luigi class CheckSource(luigi.Task): date = luigi.DateParameter() def output(self): source_path = f"/data/metrics/{self.date:%Y%m%d}/raw.dat" return luigi.LocalTarget(source_path) def run(self): if self.output().exists(): self.output().touch()

这里有个细节：check任务的output不能直接指向源文件。因为target存在的本意是“任务产出存在”，如果把源文件当作output，Luigi会认为任务自己创建了源文件，语义上不干净，而且重跑时可能出现误判。我早期版本直接返回源文件target，结果调度器认为source任务“已完成”，实际是文件早就存在。后来改成check任务内部touch自己的标记文件，才避免了这个问题。

GenerateRows在run里读源文件，转换后写出中间文件：

class GenerateRows(luigi.Task): date = luigi.DateParameter() def requires(self): return CheckSource(date=self.date) def output(self): return luigi.LocalTarget(f"/data/metrics/{self.date:%Y%m%d}/rows.json") def run(self): with self.input().open('r') as src: rows = [parse_line(line) for line in src] with self.output().open('w') as out: json.dump(rows, out)

self.input()会自动对接requires返回的target，读源文件就变成一行代码。这也是我比较喜欢Luigi的地方：依赖和输入输出是同一个对象，不容易搞乱路径。

3.4 写入HBase：批量Put与batch

WriteHBase是核心，我的做法是分块读中间文件，每5000行一个batch，用happybase的batch接口写：

import json import luigi import happybase class WriteHBase(luigi.Task): date = luigi.DateParameter() def requires(self): return GenerateRows(date=self.date) def output(self): return HBaseMarkTarget(date=self.date) def run(self): conn = happybase.Connection('thrift-host', port=9090, timeout=30000) table = conn.table('metric_data') with self.input().open('r') as f: rows = json.load(f) with table.batch(batch_size=5000, transaction=False) as batch: for r in rows: row_key = f"{self.date:%Y%m%d}_{r['merchant_id']}_{r['metric_name']}" batch.put(row_key, {b'cf:value': str(r['value']).encode()}) conn.close()

这里有两个坑值得写出来。第一，row_key的拼接顺序要保证日期在前、商户ID其次，这样同一日期的数据在HBase里物理连续，Scan按日期前缀就能扫到。如果ID在前日期在后，同一天的记录会散落很多region，扫描性能会差到没法看。第二，batch_size不是越大越好，实测里5000行左右比较稳，超过一两万行时，一次RPC耗时上升明显，反而拖慢整体速度。

transaction=False表示不用行级事务，批任务里一般够用，还能减少跨行事务开销。如果业务要求强一致性的批量写入，才需要另做方案。

3.5 让Luigi跑起来：luigi命令行

任务全部定义好后，在终端执行：

luigid --port 8082 --background luigi --module my_pipeline WriteHBase --date 2025-01-01 --workers 2

第一行启动Luigi调度器，第二行提交当天任务。Luigi会自动分析依赖，从上到下执行CheckSource、GenerateRows、WriteHBase。因为有调度器，你在浏览器打开8082端口能看到任务状态和依赖图，不需要额外装监控组件。

这里有个容易忽略的点：luigid的state文件要放在所有worker都能访问的地方。如果跨机器执行任务，调度器的数据库路径不一致，任务状态就会对不上。建议在配置里显式指定state文件路径，不要用默认值。

4. 进阶：读HBase做Scan，和Luigi的Target输出管理

4.1 Scan设计：RowKey范围、过滤器与批量迭代

写完数据之后，通常需要扫描出来做校验或者下游消费。happybase的scan支持row_start、row_stop指定行键范围，也支持filter字符串。我的行键格式是日期_商户ID_指标名，那么扫描当天全量数据只需要：

for row_key, data in table.scan(row_start=b'20250101_', row_stop=b'20250102_'): value = data.get(b'cf:value')

row_start传20250101_，row_stop传20250102_，恰好把1月1日当天所有行划进来。这个设计的价值在于：不需要全表扫描，HBase内部按字典序快速定位到起始行，然后顺序迭代。在亿级行表里，这个Scan也只触及当天数据对应的region。

scan还有一个batch参数。如果你不设batch，一次RPC可能返回几万行，内存压力大不算，还容易触发服务端超时。我的经验是batch设成1000行比较稳，配合scan_batch在服务端每次返回一组，客户端迭代完再拉下一组。

注意：Scan时不要拿一个大列表把所有结果装进去，除非你确认当天行数很少。正确姿势是迭代器模式，处理一行丢一行，内存保持稳定。

4.2 用自定义Target标记一次HBase导出

前面WriteHBase的output我写的是HBaseMarkTarget，这个类需要自己实现。Luigi自带的Target主要是文件型LocalTarget和HdfsTarget，但HBase任务的“产出”并不是一个文件，而是“某个时间段的数据已写入”。我们可以用一张单独的标记表或标记行来模拟Target。

我这里定义一个简单版本，把标记写到本地文件，key里包含日期：

class HBaseMarkTarget(luigi.LocalTarget): def __init__(self, date): super().__init__(f"/data/metrics/{date:%Y%m%d}/hbase_mark")

这样WriteHBase只要output文件存在，Luigi就认为任务完成。缺点是没法跟HBase真实数据联动。更严谨的做法是，在HBase里创建一个专用标记列族，写入完成后再put一行状态，并让自定义Target去读这个状态。写起来会麻烦一点，但一致性更好。

我个人的实际偏好是：如果任务量不大，用文件标记够用；如果任务量大、经常要核验HBase真实状态，就做HBase状态行。后面讲一致性问题时你会理解为什么。

4.3 原子性、重跑与幂等：Target为什么是核心

Luigi对重跑的处理逻辑很简单：你的Task如果output不存在，就重新run；如果output存在，默认就跳过。这个逻辑靠的是Target的exists()。

放到HBase场景里就有个微妙问题：WriteHBase往HBase写了几百万行，最后写了标记文件。如果写入中途失败，没写标记文件，重跑没问题，重新put一遍也没有重复风险吗？不完全没风险。假如中途失败但部分batch已经成功提交，重跑会再次put这些行。如果row_key设计一样，值也一样，HBase里其实是覆盖更新，不会出现两行重复；但如果是“值变化了”或者“同一row_key被两个不同任务写”，就可能出现互相覆盖。

所以我的经验是：每个任务的row_key要保证全局稳定，而且写入必须是全量覆盖语义，而不是追加语义。在此基础上，重跑一百次也安全。如果你的业务确实是累加型数据，那就得加分区前缀或时间戳后缀，避免重跑造成数据翻倍。

4.4 清理与TTL设计

HBase表默认是不删除数据的，随着任务日复一日跑，数据只会膨胀。最简单的方案是在建表时设置TTL，比如保留30天：

conn.create_table('metric_data', {'cf': {'max_versions': 1, 'time_to_live': 2592000}})

TTL是HBase在读取和Compaction时清理过期数据，不需要在Python层额外写删除任务。但也别完全依赖TTL，因为TTL清理有滞后，而且对rowkey设计有影响，过期时间是从写入时间开始算的。如果你做历史数据重跑，要小心：重跑30天前的数据，即使写入成功，也可能在一两天后被TTL清掉，业务侧会认为数据缺失。

所以这种场景的建议是：把“任务窗口期”和“数据保留周期”分开设计。重跑只允许近几天，历史补数走另一条显式清理加重新导入的流程，不要直接复用每日批任务。

5. 依赖管理、失败重试和资源控制

5.1 多个上游依赖的表示

现实里的任务往往不是单链依赖。比如GenerateRows可能需要CheckSource和CheckDict两个上游；WriteHBase可能需要GenerateRows和CheckTableReady两个上游。Luigi的requires()可以返回list，也可以返回dict。

list版本在self.input()时返回一个target列表，dict版本在self.input()时返回一个dict。我习惯用dict，因为代码可读性更高：

def requires(self): return { 'rows': GenerateRows(date=self.date), 'table_status': CheckTableReady(date=self.date) } def run(self): rows_input = self.input()['rows'] status = self.input()['table_status']

一旦某个上游失败，下游不会启动。这个行为在Luigi里是自动的，不需要你写额外判断。想要判断“上游虽然失败但可以继续”的场景，Luigi会显得别扭，因为这套框架的哲学就是“失败即停”。

5.2 失败重试：retry_count与retry_delay

Luigi内置的重试机制默认不打开，需要显式设置。在Task类上定义：

class WriteHBase(luigi.Task): date = luigi.DateParameter() retry_count = 3 retry_delay = 60

retry_count限制同一任务实例最多尝试次数。如果超过了，任务整体失败，Luigi会把依赖它的后续任务标记为不可执行。这里要特别注意：如果run里已经往HBase写了部分数据，重试机制和幂等设计要配合好，否则重试次数越多，脏数据越多。我一直坚持“可重查、可覆盖”的row_key设计，就是给重试机制兜底的。

retry_delay的单位默认是秒，设为60意味着第一次失败后至少等60秒再重试。这个延迟主要用来避开瞬时的网络抖动或集群压力，不要设太短，不然失败高峰时一堆任务同时重试，容易把HBase打得更惨。

5.3 并发与资源控制

Luigi默认是单worker执行，如果任务本身写比较重，单worker依次跑就行。多任务并行可以开--workers 4。但HBase类的任务要谨慎开高并发：Thrift服务端不一定能扛住几十个客户端同时批量写，RegionServer的CPU和IO都会飙高。

我自己在批处理场景里的经验是：写入HBase的任务并发控制在2到4个；如果是Scan为主的任务，可以适当多开，因为Scan更多是IO等待，对CPU压力相对小。还有一类经验，把重IO任务和轻任务分开调度，不要开一个workers=10把所有任务一起并行，不然HBase连接数会暴涨。

5.4 调度器中央状态：luigid与任务历史

Luigi的中央调度器服务叫luigid，启动之后任务状态和依赖关系都会记录到它的数据库里。默认情况下，luigid会把state存储到本地文件。跑任务时如果不开luigid，任务会直接执行但看不到历史状态，所以要用Luigi做正式调度，luigid基本是必须的。

luigid还提供了一套简单的REST接口，你可以用它查询任务状态，做监控告警。比如每天写完后查一下VerifyCount是否处于DONE状态，不DONE就告警。这比守着日志文件靠谱得多。真正落地的时候，我还会配合一个简单的Webhook通知，在任务图全部结束时发一条消息，不管是成功还是失败。

6. 实战踩坑：几个真实问题的排查过程

6.1 Thrift连接被断：socket异常导致任务中断

某次跑一个全量重导任务，Scan阶段跑到一半，进程抛了ConnectionResetError。查日志发现，HBase Thrift服务端对长时间空闲或超时未返回的连接做了断开处理，而客户端在异常发生后没有重建连接，任务就直接失败。

解决方案分两步。第一步，catch住异常后，重新创建Connection对象，不能复用旧连接；第二步，在scan循环里用try包裹单次迭代，如果出现网络异常就重建连接并再次发起scan，注意用一个游标记录上次处理到的row_key，从断点继续，而不是从头开始。

这个“断点续扫”逻辑我在多个项目里写过，核心就一句话：记住上一个处理成功的row_key，异常后从它之后重新scan。HBase的scan本身有状态，断开之后重开扫描必然从头开始，但只要不重新处理断点之前的行，就不会重复消费。

6.2 RowKey热写入：预分区很重要

刚开始建的HBase表只有默认一个region，写入时所有Put请求全压在一个RegionServer上，写入延迟飙升，批次越写越慢。后来看HBase的监控，发现热点集中在单个节点，才意识到问题出在预分区。

解决办法是建表时预分配region。比如用日期加商户ID做前缀，已知商户ID哈希范围，可以在建表时指定splits：

splits = [b'2025-01-01', b'2025-02-01', b'2025-03-01', b'2025-04-01'] conn.create_table('metric_data', {'cf': {}}, splits=splits)

这个方案适合按时间分区的场景。如果是随机ID，用HexStringSplit预分区工具更合适。预分区不是灵丹妙药，但能解决绝大多数“全表单region”的问题。数据量变大之后，HBase自己会再split，不过主动预分区可以让写入从一开始就分散到多个节点。

6.3 任务状态与HBase数据的一致性问题

有一类问题特别容易踩：WriteHBase跑完，Luigi把它标成DONE，但是第二天你发现当天HBase数据缺了一部分。查来查去，原因不是Luigi判断错了，而是我的output标记文件写在了所有HBase操作之前，或者说run里是先touch了标记文件再写数据。

Luigi判断任务是否完成，只看output的exists。如果你在代码里先创建或touch了标记，后续写HBase失败，Luigi可能已经认为任务完成了，这样重跑就永远不会发生。正确的顺序是：run里先写HBase，确保全部成功后再创建标记文件。我把这个原则记成一句话：target是用来证明“任务已经完成”，不是用来提示“任务快完成”。

6.4 重跑导致数据重复：幂等设计再强调

我再讲一个更隐蔽的案例。某次手动改了源文件，要求历史日期重跑。由于row_key里没有版本信息，重跑时同一row_key的新值覆盖了旧值，看起来挺好。但下游有个任务是把当天每个商户的多个指标汇总，它读取的是HBase的历史数据，却发现有些商户的汇总值是旧的，有些是新的。

这不是写入重复，而是“重跑只覆盖了部分数据”。因为重跑过程中任务失败了一次，第一次写的部分行已经提交，第二次重跑又写了一批，但同一row_key的两次写入都成功，问题在于batch提交顺序和row_key顺序不完全一致，导致最后存在某些行是旧值、某些行是新值。解决办法很简单：不要在一个Task里用多个batch分段写同一批目标行，而是保持整个重跑过程里同一row_key的写入是最后一次为准，并且保证整批要么全更新，要么不更新。对于这种场景，我会在写入前先把当天前缀的整片区删除，再重新写入，用“先删后插”换取强一致。

最后再分享一个小技巧：第一次在项目里集成Luigi时，不要一上来就追求完整的可视化、多worker、分布式调度，先把一个最简单的单链依赖跑通，确认HBase写入和标记文件逻辑没问题，再逐步加并发和重试。我用这个方式带过好几个刚开始接触Luigi的同事，基本没出过大乱子。编排框架本身不复杂，复杂的往往是数据，而数据的问题，越早暴露越好。