|
@@ -50,9 +50,6 @@ def get_table_data(project, table, partition):
|
|
|
"""获取全部分区数据"""
|
|
"""获取全部分区数据"""
|
|
|
|
|
|
|
|
records = get_data_from_odps(date=partition, project=project, table=table)
|
|
records = get_data_from_odps(date=partition, project=project, table=table)
|
|
|
- print(records)
|
|
|
|
|
- print(len(records))
|
|
|
|
|
- print(type(records))
|
|
|
|
|
|
|
|
|
|
data = []
|
|
data = []
|
|
|
for record in records:
|
|
for record in records:
|
|
@@ -95,7 +92,7 @@ def h_timer_check():
|
|
|
log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt))
|
|
log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt))
|
|
|
#2 读取数据表
|
|
#2 读取数据表
|
|
|
data = get_table_data(project, table, partition)
|
|
data = get_table_data(project, table, partition)
|
|
|
- log_.info("数据处理完成，开始处理和写入redis。")
|
|
|
|
|
|
|
+ log_.info("数据处理完成，数据数量={}，开始处理和写入redis。".format(len(data)))
|
|
|
#3 写入redis
|
|
#3 写入redis
|
|
|
records_process(data, process_and_store, max_size=10, num_workers=5)
|
|
records_process(data, process_and_store, max_size=10, num_workers=5)
|
|
|
|
|
|
zhangbo