HdfsImporter

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

1. HdfsImporter 概述

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

2. 运行环境

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

3. 经典使用方法

一次新启动的导入步骤如下：

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

2. 切换到 sa_cluster 账户：sudo su - sa_cluster

3. 请确定 HDFS 上数据不再进行修改（增删修改文件）后 ，运行 HdfsImporter

使用方式 ：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh \
--path /data/input \
--project default

备注：

• 自 1.10 版本起，直接运行 jar 包的方式不再被支持。

• 自 1.13 版本起，可直接用简略的 hdfs_importer 命令替换 sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh

• 导入后清理数据较复杂，请检查好再操作。对同一份数据多次运行导入会导致数据重复。

• path 可以是一个包含子目录的数据目录。

• HdfsImporter 执行成功之后，数据大约会在 1分钟之后 被查询到。

• hdfs 操作命令文档参考

4. 配置项说明

列出所有配置项：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh --help

参数	必填	说明	默认值
path	是	需要导入的数据所在的 HDFS 路径。
project	否	期望导入的 project 名称，不指定的则导入到 default。注：导入数据中的 project 字段会失效，以此参数指定为准。	default
all_data_without_track_signup	否	是否所有数据（包括之前各种类型导入）都不包含 track_signup 类型数据。若所有导入数据均不包含 track_signup 类型数据，添加这个选项可以提高导入速度。注意，使用过用户关联功能则不能使用该选项。
split_max_size_mb	否	MapReduce 每个 Mapper 处理分片的最大值。该值会影响 Mapper 数量。	512
job_queue_name	否	MapReduce Job 的队列名。
split_reduce_size_mb	否	MapReduce 每个 Reducer 处理的数据量大小。该值会影响 Reducer 的数据。Reducer 个数 = 输入的总数据量 / split_reduce_size_mb 。	2048
mapper_max_memory_size_mb	否	MapReduce 每个 Mapper 使用的内存最大值。	1024
reduce_max_memory_size_mb	否	MapReduce 每个 Reducer 使用的内存最大值。	2048
reduce_output_file_size_mb	否	MapReduce 每个 Reducer 输出的文件最大值。	256
shuffle_partition_split_size	否	Shuffer 时，每个 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

一个复杂的例子：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh \
--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

5. 查询导入历史

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

使用方式 ：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh list \
--project default

列出所有配置项 ：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh list --help

参数	必填	说明	默认值
project_name	否	project name, 可指定按照 project name 查询导入任务记录
session_id	否	session id, 可指定按照 session id 查询导入任务记录
job_name	否	job name，可指定按照 job_name 查询导入任务记录。这里的 job_name 为导入任务中的 user_define_job_name
status	否	hdfs importer 导入任务的状态，status 值可为 WAITING，RUNNING，SUCCESS，FAILED
start_time	否	start time, 可查询startTime比传入的指定时间晚的导入记录， 格式%Y-%m-%d %H:%M:%S
max_num	否	recent N times，可指定最近N次导入记录	10
dump	否	查询结果输出到指定的FILENAME；如果没有指定，则会输出到控制台
full	否	指定后则查看所有任务，否则同一个session_id只查看最新的任务	False

例如，查询按小时的例行导入任务：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh 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

输出到文件中的结果为：

[
    {
        "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"
    }
]

再如，查询指定 session_id 的所有导入任务状态：

sh $SENSORS_ANALYTICS_HOME/tools/hdfs_importer/bin/hdfs_importer.sh list \
--session_id 306 \
--full

输出到控制台的结果为

[
    {
        "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"
    }
]

6. 更新日志

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. 新增查询导入历史子命令

2017-11-29 (SA 版本号：1.10.3329)

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

2017-07-10 (SA 版本号：1.7.2628)

1. 新增 Mapper 内存参数 mapper_max_memory_size_mb

2. 支持用户自定义 InputFormat，详情请参考 github ：HDFSImporter 自定义 InputFormat

2017-07-04 (SA 版本号：1.7.2614)

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