在使用前,请先阅读数据模型数据格式的介绍。

1. HdfsImporter 概述

HdfsImporter 与 BatchImporter 一样用于将历史数据或外部数据导入神策分析。

2. 运行环境

HdfsImporter 仅可在神策分析集群版使用,且只能运行在神策分析集群的 Hadoop 环境中。

3. 经典使用方法

3.1. 一次新启动的导入步骤如下:

  1. 将数据置于神策分析集群上的 HDFS 中。例如 HDFS 集群 /data/input 目录,其中包含文件data01、data02、data03等文件,文件内容为每行一个符合 数据格式 的 Json。
  2. 切换到 sa_cluster 账户: 

    sudo su - sa_cluster
    CODE
  3. 请确定 HDFS 上数据不再进行修改(增删修改文件)后,运行 HdfsImporter。

3.2. 使用方式

  1.  先检查数据正确性,但不进行真的数据导入:开始 debug 模式,数据只会以 `JSON` 文本的方式输出到相应的 HDFS 目录下,并且不会真正的导入到神策系统当中。

    hdfs_importer \
    --path /data/input \
    --project default \
    --debug
    BASH
  2. 检查无误后,进行真正的数据导入 :

    hdfs_importer \
    --path /data/input \
    --project default
    BASH

3.3. 备注

  • 1.13 版本以前,由于没有简略命令 hdfs_importer , 需使用 sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh 代替
  • 导入后清理数据较复杂,请检查好再操作。对同一份数据多次运行导入会导致数据重复。
  • path 可以是一个包含子目录的数据目录。
  • HdfsImporter 执行成功之后,数据大约会在1分钟之后被查询到。
  • hdfs 操作命令文档参考

4. 配置项说明

  • 列出所有配置项:

    hdfs_importer --help
    CODE
    参数必填说明默认值
    path需要导入的数据所在的 HDFS 路径。
    project期望导入的 project 名称,不指定的则导入到 default。注:导入数据中的 project 字段会失效,以此参数指定为准。default
    all_data_without_track_signup是否所有数据(包括之前各种类型导入)都不包含 track_signup 类型数据。若所有导入数据均不包含 track_signup 类型数据,添加这个选项可以提高导入速度。注意,导入项目中使用过用户关联功能则不能使用该选项。
    split_max_size_mbMapReduce 每个 Mapper 处理分片的最大值。该值会影响 Mapper 数量。512
    job_queue_nameMapReduce Job 的队列名。
    split_reduce_size_mbMapReduce 每个 Reducer 处理的数据量大小。该值会影响 Reducer 的数据。Reducer 个数 = 输入的总数据量 / split_reduce_size_mb 。2048
    mapper_max_memory_size_mbMapReduce 每个 Mapper 使用的内存最大值。1024
    reduce_max_memory_size_mbMapReduce 每个 Reducer 使用的内存最大值。2048
    reduce_output_file_size_mbMapReduce 每个 Reducer 输出的文件最大值。256
    shuffle_partition_split_sizeShuffer 时,每个 Mapper 切分 partition 的大小,一般用来调整数据分布不均匀500000
    expired_record_filter_after_hour允许导入的数据时间区间截至未来的小时数,默认为 1,即超过未来 1 小时的数据将被过滤。1
    expired_record_filter_before_hour允许导入的数据时间区间向前的小时数,默认为 17520,即超过 2 年以前的数据将被过滤。17520
    user_define_job_name允许自定义 job name,比如在 HdfsImpoter-12345-[your_name] 中,your_name 是自定义的部分
    debug开始 debug 模式,数据只会以 JSON 文本的方式输出到相应的 HDFS 目录下,并且不会真正的导入到神策系统当中。
    disable_import_succeeded_source_path禁止重复导入已成功过的导入路径false
    write_import_info在命令执行路径生成 import info 文件false
  • 一个复杂的例子:

    hdfs_importer \
    --path /data/input \
    --project test_project \
    --all_data_without_track_signup \
    --split_max_size_mb 128 \
    --job_queue_name data_import_queue \
    --split_reduce_size_mb 2048 \
    --reduce_max_memory_size_mb 2048 \
    --reduce_output_file_size_mb 256 \
    --disable_import_succeeded_source_path
    CODE

5. 查询导入历史

自1.13 版本,支持查询 hdfs importer 导入历史, 查询结果以 JSON 格式输出。

  • 使用方式 :

    hdfs_importer list \
    --project default
    BASH
  • 列出所有配置项 :

    hdfs_importer list --help
    BASH
    参数 必填说明 默认值
    project_nameproject name, 可指定按照 project name 查询导入任务记录
    session_idsession id, 可指定按照 session id 查询导入任务记录
    job_namejob name,可指定按照 job_name 查询导入任务记录。这里的 job_name 为导入任务中的 user_define_job_name
    statushdfs importer 导入任务的状态,status 值可为 WAITING,RUNNING,SUCCESS,FAILED
    start_timestart time, 可查询startTime比传入的指定时间晚的导入记录, 格式%Y-%m-%d %H:%M:%S
    max_numrecent N times,可指定最近N次导入记录10
    dump查询结果输出到指定的FILENAME;如果没有指定,则会输出到控制台
    full指定后则查看所有任务,否则同一个session_id只查看最新的任务False
  • 例如,查询按小时的例行导入任务: 

    hdfs_importer list \
    --project_name test_project \
    --job_name hourly_import \
    --status SUCCESS \
    --start_time '2018-08-30 00:00:00' \
    --max_num 2 \
    --dump /data/work/list_output_file1
    BASH

    输出到文件中的结果为:

    [
        {
            "id": 12,
            "session_id": 320,
            "job_name": "hourly_import",
            "scheduler_job_record_id": null,
            "import_path": "/sa/tmp/hdfsImport",
            "parameters": "{\"path\":\"/sa/tmp/hdfsImport\",\"project\":\"test_project\",\"user_define_job_name\":\"hourly_import\"}",
            "start_time": "2018-09-11 18:46:50",
            "end_time": "2018-09-11 18:49:46",
            "counter_info": "{\"event_read_count\":252600,\"event_import_count\":252600,\"event_skipped_count\":0,\"profile_read_count\":10104,\"profile_import_count\":10104,\"profile_skipped_count\":0}",
            "log_path": "/data/sa_cluster/logs/tools/hdfsimporter/hdfsimporter-43c7d4ea-0b14-48f6-8b03-764178e927ae.log",
            "event_job_id": "job_1536029076029_3074",
            "profile_job_id": "job_1536029076029_3077",
            "event_job_status": "SUCCESS",
            "profile_job_status": "SUCCESS",
            "event_data_load_status": "SUCCESS",
            "project_name": "test_project"
        },
        {
            "id": 10,
            "session_id": 317,
            "job_name": "hourly_import",
            "scheduler_job_record_id": null,
            "import_path": "/sa/tmp/hdfsImport",
            "parameters": "{\"path\":\"/sa/tmp/hdfsImport\",\"project\":\"test_project\",\"user_define_job_name\":\"hourly_import\"}",
            "start_time": "2018-09-11 10:23:20",
            "end_time": "2018-09-11 10:26:21",
            "counter_info": "{\"event_read_count\":252600,\"event_import_count\":252600,\"event_skipped_count\":0,\"profile_read_count\":10104,\"profile_import_count\":10104,\"profile_skipped_count\":0}",
            "log_path": "/data/sa_cluster/logs/tools/hdfsimporter/hdfsimporter-67a00f94-67d8-415e-a004-c9ca82a17a2a.log",
            "event_job_id": "job_1536029076029_3044",
            "profile_job_id": null,
            "event_job_status": "SUCCESS",
            "profile_job_status": null,
            "event_data_load_status": "SUCCESS",
            "project_name": "test_project"
        }
    ]
    BASH
  • 再如,查询指定 session_id 的所有导入任务状态: 

    hdfs_importer list \
    --session_id 306 \
    --full
    BASH

    输出到控制台的结果为

    [
        {
            "id": 8,
            "session_id": 306,
            "job_name": "hourly_import",
            "scheduler_job_record_id": null,
            "import_path": "/sa/tmp/hdfsImport",
            "parameters": "{\"session\":\"HdfsImport-306\"}",
            "start_time": "2018-09-10 19:02:08",
            "end_time": "2018-09-10 19:02:46",
            "counter_info": "{\"profile_read_count\":10104,\"profile_import_count\":10104,\"profile_skipped_count\":0}",
            "log_path": "/data/sa_cluster/logs/tools/hdfsimporter/hdfsimporter-0634112a-6b90-40db-a26d-5492dbc7b995.log",
            "event_job_id": null,
            "profile_job_id": "job_1536029076029_3084",
            "event_job_status": null,
            "profile_job_status": "SUCCESS",
            "event_data_load_status": null,
            "project_name": "test_project"
        },
        {
            "id": 7,
            "session_id": 306,
            "job_name": "hourly_import",
            "scheduler_job_record_id": null,
            "import_path": "/sa/tmp/hdfsImport",
            "parameters": "{\"path\":\"/sa/tmp/hdfsImport\",\"project\":\"test_project\",\"user_define_job_name\":\"hourly_import\"}",
            "start_time": "2018-09-10 18:58:45",
            "end_time": "2018-09-10 19:01:10",
            "counter_info": "{\"event_read_count\":252600,\"event_import_count\":252600,\"event_skipped_count\":0}",
            "log_path": "/data/sa_cluster/logs/tools/hdfsimporter/hdfsimporter-c7514335-5a55-42b8-bfd3-0ad7a27ec1a3.log",
            "event_job_id": "job_1536029076029_3082",
            "profile_job_id": "job_1536029076029_3083",
            "event_job_status": "SUCCESS",
            "profile_job_status": "FAILED",
            "event_data_load_status": "SUCCESS",
            "project_name": "test_project"
        }
    ]
    BASH


6. 更新日志

6.1. 2018-09-13 (SA 版本号:1.13)

  1. 导入配置项中增加参数 disable_import_succeeded_source_path
  2. 导入配置项中增加参数 write_import_info
  3. 可使用简略的 hdfs_importer 命令替换 sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh
  4. 新增查询导入历史子命令

6.2. 2017-11-29 (SA 版本号:1.10.3329)

  1. 支持自定义 job_name 参数: user_define_job_name

6.3. 2017-07-10 (SA 版本号:1.7.2628)

  1. 新增 Mapper 内存参数 mapper_max_memory_size_mb
  2. 支持用户自定义 InputFormat,详情请参考 github :HDFSImporter 自定义 InputFormat

6.4. 2017-07-04 (SA 版本号:1.7.2614)

  1. 支持用户自定义预处理模块,详情请参考 github :HDFSImporter 预处理模块
  2. 新增 debug 模式