DAG:ods-mobile_refresh_dailyschedule每天post
f1f2f3 f4
是
flinkhermes/tree/master/druidetl
kafkalog-dataquality-event
hdfs/hive
catch
A/B分流灰度测试
不做处理
/camus/topics//origin-mobile-event/dt=$dt/hour=$hour/\"
write
run_camuscamus处理/usr/local/datacenter/camus/conf/camus_origin-mobile-event.properties
获取上次成功时间戳
1 删除/datacenter/mobile_event/dt=$dt/hour=$hour/\"2 mr任务 com.mafengwo.databus.tidyMerge.Bootstrap 输入目录 /camus/topics /datacenter//origin-mobile- event/dt=$dt/hour=$hour/ 输出目录 /datacenter/mobile_event/dt=$dt/hour=$hour/3 新建临时文件 /datacenter/mobile_event/dt=$dt/hour=$hour/done_${dt}_${hour}
数据对量平台
flinkdatapipeline项目涉及降级策略
数据源esdruidredis
run_mergecom.mafengwo.databus.tidyMerge.Bootstrap
android_oeid0 (ime0) 001span style=\"font-size: inherit;\
成功
es
业务kafka
5分钟执行一次
1 替换 $_COOKIE $_POST $_GET中__open_udid 的值2 cookie 赋值__s_open_udid
curd
查询tidb表
写入临时分钟级文件
1 删除 /tmp/mobile_event/dt=$dt/hour=$hour2 mr任务 com.mafengwo.databus.mobile_code.parquet.Bootstrap /datacenter/mobile_event/dt=$dt/hour=$hour/ 输入 /tmp/mobile_event/dt=$dt/hour=$hour 输出 3 hive parquet 入库
临时文件只存时间戳
dataquality-event
druid
data-es-number实时
run_parquet
遍历读取每个文件的 行数/记录数
data-pipeline-drop实时
Tranquilty项目http://192.168.7.225:8200/v1/post
记录错误日志
t1
获取每个文件1 行数-日志条数2 文件绝对路径3 hostname
写入时间戳
替换mobile_event中open_udid
16:05 - 15:45 15:4*16:15 - 15:55 15:5*16:25 - 16:05 16:0*16:35 - 16:15 16:1*16:45 - 16:25 16:2*16:55 - 16:35 16:3*17:05 - 16:45 16:4*17:15 - 16:55 16:5*17:25 - 17:05 17:0*
接受数据发送到service
日志文件成功/失败
获取上次执行时间戳
device_mapping_id 为空 | | 长度 < 8
开始
整个文件内容
日志文件清理
否
t0
logdataquality脚本调度
data-source-number离线
写入
线程1cron每5分钟起一次 守护模式/cron模式根据系统部部署可灵活变动
t-20min 以前读取分钟级(十位数)文件list
内存/文件 ?
做数据插入
客户端请求
生成上次成功时间戳到1分钟以前的文件list
service
写文件
根据客户端传过来的参数拼装device_mapping_id
是否和上报open_udid一致
post
data-redis-number实时
判断数据流中open_udid长度是否合法>7 && <45
airflow批处理
data-druid-number实时
执行
json统计数据
mobile_event 事件数据流
response
读取
数据kafka
camuscamus-confluent项目
ods-mobile_event_etl 小时刷新流程
operation-monitor项目collect.pyawk 统计系统部日志文件
s1
失败/超时
data-camus-hour离线
redis
firebeat
s0
