大数据测试中一些工具使用

一.数据处理架构

    如图，数据流转主要有两条线，实时计算流程和离线计算流程

• 实时计算：事件(hive表)----(使用dw-event-to-collector.sh发送事件)---->收数工具collector-------->flume分发-------->kafka缓存-------->flink计算-------->hbase-------->elasticsearch
• 离线计算：事件(hive表)----(主动读hive表)---->hdfs-------->flink计算-------->hbase-------->elasticsearch

二.实时计算过程中工具使用

  1.hive

• 进入hive数仓: hive
• 查看当前数据库: show databases;
• 切换到cdp库: use cdp;
• 创建表(SMH前端的导出事件配置中有自动生成的语句):  
  CREATE TABLE IF NOT EXISTS tablename( 
  uid string, 
  event_time bigint, 
  touch_point_id string
  )partitioned by (process_date string) 
  ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY ' ' 
  LINES TERMINATED BY ' ' 
  STORED AS TEXTFILE;

• 查看建表命令: show create table c8_shopping;
• 查看当前表: show tables;
• 查看表中列名: desc tablename;
• 把事件插入对应hive表中:  load data local inpath "/home/hadoop/shopping.txt" into table tablename partition(process_date="2019-07-22");
• 查询表中数据: select * from tablename where process_date = '2019-04-26' limit 10;
• 查询前执行该命令列名和数据一起显示:  set hive.cli.print.header=true;
• 删除表中数据: truncate table tablename;
• 删除表: drop table tablename;

  2.kafka

   查询kafka消费情况，路径：/home/hadoop/kafka_2.11-0.10.2.0/bin

   命令： sh kafka-console-consumer.sh --topic event_c8 --from-beginning --bootstrap-server 172.00.0.000:9092 > event_c8

  3.flink

• 重启flink任务，路径：/home/hadoop/cdp-etl-jobs/bin/job/realtime
• 关闭flink任务：yarn application -kill 任务id
• 启动flink任务：sh indexing-trait.sh        sh calculate-trait.sh

  4.hbase

• 进入hbase：hbase shell
• 查看已存在的表：list
• 查询某特性的值：scan 'trait_c8',{COLUMNS=>['d:t1425','d:uid']}
• 查询某uid删除状态：scan 'trait_c8', {COLUMNS => 'd:delete_status',FILTER => "ValueFilter(=,'substring:true')"}
• 查询某个uid： get 'trait_c8','fff144eb653e7348f051307cde7db169'
• 删除表中数据：truncate "tablename";         flush "tablename";
• 删除表：disable table;          drop table;
• hbase全量同步到es：cdp/cdp-etl-jobs/bin/job/batch/trait-crowd-calc.sh -calcType sync     增量为：incr

  5.elasticsearch

   查询工具可以使用kibana或者elasticsearch head插件，常用命令：

• 查询特性：
  GET /trait_c39/trait_c39/_search?size=1000
  {
  "query": {
  "match_all": {}
  },
  "_source": ["t596"]
  }

• 查询人群：
  GET /trait_c39/trait_c39/_search?size=1000
  {
  "query": {
  "match_all": {}
  },
  "post_filter": {"term": {
  "crowds_code": "cr197"
  }}
  }

• 查询某个uid：
  GET /trait_c33/trait_c33/uid-1

三.离线计算过程中工具使用

  1.hdfs

前端页面查询地址：http://172.23.x.xxx:50070/explorer.html#/cdp/warehouse

查看目录：hadoop fs -ls /cdp/warehouse/c8/offline/

查看文件：hadoop fs -cat /cdp/warehouse/c8/offline/shopping.txt

下载数据：hadoop fs -get /cdp/warehouse/c8/offline/

删除文件：hadoop fs -rm -r /cdp/warehouse/c8/offline/shopping.txt

  2.azkaban

• cdp-batch-process  离线批处理数据
  dw-etl-process   数仓etl开始
  dw-event-to-hdfs  主动读取事件导入到hdfs
  user-delete  用户删除
  event-ub-to-hbase  发送事件到hbase，用于用户档案数据展示
  common-jobs-config  生成作业配置信息,地址:/home/hadoop/cdp-etl-jobs/jobs-tmp/codes/
      ALL_EVENT_TRAIT  所有事件到达时触发的特性list
      ALL_ACC_TRAIT  除timeline外,所有事件触发类的累加特性list
      ALL_REF_TRAIT  所有特性变化触发的特性list
      ALL_CROWD  渠道内全量人群list
      CALC_EVENT_TRAIT  事件到达时触发且需重新计算的特性list
      CALC_TRAIT  特性变化时触发且需重新计算的特性list
      CALC_CROWD  当天需要计算的人群list,包括重新计算的人群,符合周期的人群
      CLEAN_CROWD  需删除的人群list 
      CLEAN_TRAIT  需删除的特性list
      EXPORT_TRAIT idmapping时需要导出的特性list
      CANCELED_TRAIT 取消授权影响的特性list
  event-trait-calc-full  全量重跑数据,traitupdate判断历史最新数据赋值给特性
  event-trait-calc-incr  计算每日数仓增量的数据,traitupdate只发送当天的数据
  event-trait-calc-init  对事件到达时触发特性的重新计算,traitupdate只发送当天的数据
  trait-crowd-calc  计算人群,对特性变化时触发的特性重新计算，timeline类型特性,更新节点管理员/运营专员数据
  id-mapping-clean  删除作废的mapping关系
  id-mapping-init  idMapping初始化和建立mapping关系
  id-mapping-copy  idMapping后的特性拷贝
  report-crowd-count  更新人群数量到mysql，cdp_crowd表crowd_scale列
  report-metric  定时计算所有人群长期跟踪指标及全渠道的看板指标
  cdp-batch-process
• cdp-clean-jobs  清除临时文件，人群导出过期文件
• crowd-export  人群导出
• init-channels  初始化渠道
• trait-import  特性导入