离线数仓（四）【数仓数据同步策略】（1）https://developer.aliyun.com/article/1532381
2.2.2 数据同步策略选择
两种策略都能保证数据仓库和业务数据库的数据同步，那应该如何选择呢？下面对两种策略进行简要对比。
同步策略 |
优点 |
缺点 |
全量同步 |
逻辑简单 |
在某些情况下效率较低（比如我的一张表大小10亿条，但是我每天只增加一条）。例如某张表数据量较大，但是每天数据的变化比例很低，若对其采用每日全量同步，则会重复同步和存储大量相同的数据。 |
增量同步 |
效率高，无需同步和存储重复数据 |
逻辑复杂，需要将每日的新增及变化数据同原来的数据进行整合，才能使用 |
根据上述对比，可以得出以下结论：
通常情况，业务表数据量比较大，优先考虑增量，数据量比较小，优先考虑全量；具体选择由数仓模型决定。
| 大表 | 变化多 | 全量 |
| 大表 | 变化少 | 增量 |
| 小表（比如省份表） | 变化多 | 全量 |
| 小表 | 变化少 | 全量 |
我们一般把全量同步的表叫做维度表，把增量同步的表叫做事实表。
2.2.3 数据同步工具概述
数据同步工具种类繁多，大致可分为两类，一类是以 DataX、Sqoop 为代表的基于Select查询的离线、批量同步工具，另一类是以 Maxwell、Canal 为代表的基于数据库数据变更日志（例如MySQL 的 binlog，其会实时记录所有的 insert、update 以及 delete操作）的实时流式同步工具。
全量同步通常使用 DataX、Sqoop 等基于查询的离线同步工具。而增量同步既可以使用DataX、Sqoop等工具，也可使用 Maxwell、Canal 等工具，下面对增量同步不同方案进行简要对比。
增量同步方案 |
DataX/Sqoop |
Maxwell/Canal |
对数据库的要求 |
原理是基于查询，故若想通过select查询获取新增及变化数据，就要求数据表中存在create_time、update_time等字段，然后根据这些字段获取变更数据。 |
要求数据库记录变更操作，例如MySQL需开启binlog。 |
数据的中间状态 |
由于是离线批量同步，故若一条数据在一天中变化多次，该方案只能获取最后一个状态，中间状态无法获取。 |
由于是实时获取所有的数据变更操作，所以可以获取变更数据的所有中间状态。 |
接下来我们选择用 DataX 来做全量数据的同步工作，用 Maxwell 来做增量数据的同步工作。
2.2.4、全量数据同步
全量表数据由 DataX 从 MySQL 业务数据库直接同步到 HDFS，具体数据流向如下图所示：
1. DataX 配置文件
回顾我们执行 DataX 脚本的命令：
python /opt/module/datax/bin/datax.py /opt/module/datax/job/job.json
我们是通过把配置写进一个 json 文件然后执行的，所以这里，我们需要全量同步的表共 15 张，也就意味着需要写 15 个 json 配置文件，但是毕竟开发中不可能 100个、1000个表我们也都一个个手写，所以这里我们通过一个 python 来自动生成：
vim ~/bin/gen_import_config.py
# ecoding=utf-8 import json import getopt import os import sys import MySQLdb #MySQL相关配置，需根据实际情况作出修改 mysql_host = "hadoop102" mysql_port = "3306" mysql_user = "root" mysql_passwd = "123456" #HDFS NameNode相关配置，需根据实际情况作出修改 hdfs_nn_host = "hadoop102" hdfs_nn_port = "8020" #生成配置文件的目标路径，可根据实际情况作出修改 output_path = "/opt/module/datax/job/import" def get_connection(): return MySQLdb.connect(host=mysql_host, port=int(mysql_port), user=mysql_user, passwd=mysql_passwd) def get_mysql_meta(database, table): connection = get_connection() cursor = connection.cursor() sql = "SELECT COLUMN_NAME,DATA_TYPE from information_schema.COLUMNS WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION" cursor.execute(sql, [database, table]) fetchall = cursor.fetchall() cursor.close() connection.close() return fetchall def get_mysql_columns(database, table): return map(lambda x: x[0], get_mysql_meta(database, table)) def get_hive_columns(database, table): def type_mapping(mysql_type): mappings = { "bigint": "bigint", "int": "bigint", "smallint": "bigint", "tinyint": "bigint", "decimal": "string", "double": "double", "float": "float", "binary": "string", "char": "string", "varchar": "string", "datetime": "string", "time": "string", "timestamp": "string", "date": "string", "text": "string" } return mappings[mysql_type] meta = get_mysql_meta(database, table) return map(lambda x: {"name": x[0], "type": type_mapping(x[1].lower())}, meta) def generate_json(source_database, source_table): job = { "job": { "setting": { "speed": { "channel": 3 }, "errorLimit": { "record": 0, "percentage": 0.02 } }, "content": [{ "reader": { "name": "mysqlreader", "parameter": { "username": mysql_user, "password": mysql_passwd, "column": get_mysql_columns(source_database, source_table), "splitPk": "", "connection": [{ "table": [source_table], "jdbcUrl": ["jdbc:mysql://" + mysql_host + ":" + mysql_port + "/" + source_database] }] } }, "writer": { "name": "hdfswriter", "parameter": { "defaultFS": "hdfs://" + hdfs_nn_host + ":" + hdfs_nn_port, "fileType": "text", "path": "${targetdir}", "fileName": source_table, "column": get_hive_columns(source_database, source_table), "writeMode": "append", "fieldDelimiter": "\t", "compress": "gzip" } } }] } } if not os.path.exists(output_path): os.makedirs(output_path) with open(os.path.join(output_path, ".".join([source_database, source_table, "json"])), "w") as f: json.dump(job, f) def main(args): source_database = "" source_table = "" options, arguments = getopt.getopt(args, '-d:-t:', ['sourcedb=', 'sourcetbl=']) for opt_name, opt_value in options: if opt_name in ('-d', '--sourcedb'): source_database = opt_value if opt_name in ('-t', '--sourcetbl'): source_table = opt_value generate_json(source_database, source_table) if __name__ == '__main__': main(sys.argv[1:])
由于需要使用Python访问Mysql数据库，所以需安装驱动，命令如下：
sudo yum install -y MySQL-python
脚本使用说明：
python gen_import_config.py -d database -t table
测试一下：
python gen_import_config.py -d gmall -t base_province
注意： 这里的 hdfswriter 的 writeMode = append，这个值的意思是即使这个目录下存在文件也继续写入。除了这个参数外，还可以设置 writeMode = nonConflict，这个值的意思是，如果发现目录下有文件则停止写入，直接报错。
可以看到，生成的 json 配置文件需要我们后期指定 datax 命令时再提供一个 targetdir 参数，也就是同步到我们 HDFS 的哪个目录下。
测试一下这个配置文件能不能用：
bin/datax.py -p"-Dtargetdir=/base_province" job/import/gmall.base_province.json
离线数仓（四）【数仓数据同步策略】（3）https://developer.aliyun.com/article/1532384
