大家好,又见面了,我是你们的朋友全栈君。...在生产环境中, 使用shell脚本完成一次etl操作
1.定义一个etl函数, 里面传入json行数据, 用json.loads加载行数据,并对行数据进行判断,如果没有行数据,或data字段没有在行数据里...ETL操作
ETL常用场景:
1.清洗nginx日志信息, 预处理日志文件(每小时将上报的日志拉取到本机,hdfs命令上传集群),并清洗存入hive
2.每小时清洗用户表信息,
3.后处理清洗商户信息,...4.清洗并合并设备状态信息,
5.每小时清洗每日设备分成, 清洗并合并积分流水表信息, 每小时清洗支付宝订单表信息等,
def etl(row_str):
result = []
try:
row...pass
retuen result
设置会话
spark = SparkSession.builder.appName("程序名" % statdate分区日期