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分钟之后 被查询到。

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 模式

Last updated

Was this helpful?