1. 任务背景
- 任务说明：公司 saas 数据分析类产品，客户需要把行为数据回传到客户指定文件系统中（oss）
- 周期：T+1
- 数据格式：parquet
- 数据范围：部分表全量，部分表增量
- 其他要求：
需要历史数据，部分应用部分周期需要全量一个文件，部分历史需要每天一个文件，新的数据 T+1 。
每个文件上传成功后需要一个状态空文件 _SUCCESS 文件
2. 任务分析
a. 分析
- 本数据平台 impala+kudu+hive 架构，impala-shll可导出 csv文件。
注：如果系统简单，要求简单，也可以选择数据存成hive表形式，直接文件可parquet文件。这里选择用 impala- shell 导出方式是客户有复杂的要求等等。
- csv 转 parquet ，并且 parquet文件需要携带 schema 信息。
- 脚本需要支持按某一时间段的每天处理。
b. 方案
（1）python 做脚本
（2）impala-shell 命令方式导出符合要求的csv文件
（3）pandas 和 pyarraow.parquet 结合操作csv
（4）发送标准文件到 OSS
欢迎关注，一起学习
3. 开发
（1）导出csv文件
impala-shell
-i {imp_host}
-d {imp_database}
-q " {sql} "
-B --output_delimiter="\\t"
--print_header
-o /data/xx.csv(2) 环境执行
os.system(cmd)(3) csv 转 parquet
import pyarrow as pa
import pyarrow.parquet as pq
csv_file = pd.read_csv(csvFile, delimiter='\t',low_memory=False)
user_schema = pa.schema([
('user_id', pa.int64()),
('name', pa.string())
])
table = pa.Table.from_pandas(csv_file,schema=user_schema)
pq.write_table(table, "/data/xxxx.parquet")(4) 发送到OSS
import oss2
auth_endpoint = 'auth_endpoint'
access_key_id = 'access_key_id'
access_key_secret = 'auth_endpoint'
bucket_name = 'bucket_name'
auth = oss2.Auth(access_key_id, access_key_secret)
bucket = oss2.Bucket(auth,auth_endpoint, bucket_name)
with open(output_file, 'rb') as file_obj:
# 上传具体数据
put_object = bucket.put_object(oss_path, file_obj)
if put_object and put_object.status == 200:
# 上传_success文件夹
put_success=bucket.put_object('{0}/_SUCCESS'.format(oss_success_path), '')
if put_success and put_success.status == 200:
print("oss data success file upload success")
else:
print("oss success data upload fail")
else:
print("oss success data upload fail")
（5）日期循环
import datetime
startTime = (datetime.datetime.now() + datetime.timedelta(days=-1)).strftime("%Y%m%d")
endTime = (datetime.datetime.now()).strftime("%Y%m%d")
if (is_today == 0):
startTime = datetime.datetime.strptime('2025-05-09 00:00:00', '%Y-%m-%d %H:%M:%S').strftime('%Y%m%d')
endTime = datetime.datetime.strptime('2025-08-17 00:00:00', '%Y-%m-%d %H:%M:%S').strftime('%Y%m%d')
while startTime < endTime:
print('逻辑处理')
startTime = (datetime.datetime.strptime(startTime, '%Y%m%d') + datetime.timedelta(days=1)).strftime(
'%Y%m%d')4. 遇到的问题
(1) out of memory
Error tokenizing data. C error: out of memory因为读取 CSV文件过大，内存放不下，设置 low_memory=False，机器内存小于文件大小，那是倒不出来的，除非分块读取
file_index = 0
for file in pd.read_csv(csvFile,
delimiter='\t',
low_memory=False,
chunksize=20000):
table = pa.Table.from_pandas(file, schema=schema)
pq.write_table(table, '/data/xxxx_{1}.parquet'.format(file_index))
file_index+=1(2) 数据类型转换问题
pyarrow.lib.ArrowTypeError:
('Expected a string or bytes dtype, got int64',
'Conversion failed for column xxxx with type int64')当数据列出现 Null 的时候，转换 int64 转化不过去，就会报错，需要转为 str 同理一样，其他数据类型也会出现这种情况，当schema定义某列为int64，32，16 .... 的时候,如果出现Null，会自定识别为 float 类型。
csv_file = pd.read_csv(csvFile, delimiter='\t',low_memory=False)
csv_file['xxxx'] = csv_file['xxxx'].astype(str)
