离线数据同步变迁

第一代-基于Hadoop体系的离线数据同步

一、背景

随着业务的发展，体系举行了微服务的差分，导致数据越来越分散，很难举行一个完整的生命周期的数据查询，对于某些业务的需求支持变得越来越难，越来越复杂，也越来越难以举行职责划分。对着业务的发展，数据量越来越大之后，为了良好的业务支持，举行了分库分表，分库分表规则五花八门，一旦脱离了业务逻辑，很难确定某一条数据在哪个库哪个表。
基于这样的问题和情况，为了满意业务需求，很自然的就想到了使用大数据服务，将业务数据归集到一起，创建完整的数据仓库，便于数据的查询。
二、数据同步架构

为了追求简朴和通用，由于自身的认识现在，选择了最标准的大数据架构，即基于Hadoop的大数据体现。整个集群采用三节点，通过CDH举行集群的部署和维护。

整个数据链路为：
通过Azkaban调用Spark应用，将数据从RDS同步到Hive，运营平台和报表体系采用Presto加速访问Hive的数据。
三、数据同步具体过程

数据同步采用Spark任务来举行，将任务打包之后，上传到Azkaban调度平台，使用Azkaban举行定时调度，完成T+1级别的数据同步工作。
数据同步代码示例:

1. object MarketMysqlToHiveEtl extends SparkHivePartitionOverwriteApplication{
4.   /**
5.    * 删除已存在的分区
6.    *
7.    * @param spark SparkSessions实例
8.    * @param date 日期
9.    * @param properties 数据库配置
10.    */
11.   def delete_partition(spark: SparkSession, properties:Properties, date: String):Unit={
12.     val odsDatabaseName = properties.getProperty("hive.datasource.ods")
13.     DropPartitionTools
14.      .dropPartitionIfExists(spark,odsDatabaseName,"ods_t_money_record","ds",date)
15.     DropPartitionTools
16.      .dropPartitionIfExists(spark,odsDatabaseName,"ods_t_account","ds",date)
17.   }
21.   /**
22.    * 抽取数据
23.    * @param spark SparkSession实例
24.    * @param properties 数据库配置
25.    * @param date 日期
26.    */
27.   def loadData(spark: SparkSession, properties:Properties, date: String): Unit ={
28.     // 删除历史数据，解决重复同步问题
29.     delete_partition(spark,properties,date)
31.     // 获取数据源配置
32.     val odsDatabaseName = properties.get("hive.datasource.ods")
33.     val dataSource = DataSourceUtils.getDataSourceProperties(FinalCode.MARKET_MYSQL_FILENAME,properties)
35.     var sql = s"select id,account_id,type,original_id,original_code,money,reason,user_type,user_id,organization_id," +
36.     s"create_time,update_time,detail,deleted,parent_id,counts,'${date}' AS ds from TABLENAME where date(update_time) ='${date}'"
38.     // 同步数据
39.     MysqlToHiveTools.readFromMysqlIncrement(spark,dataSource,sql.replace("TABLENAME","t_money_record"),
40.                                             s"${odsDatabaseName}.ods_t_money_record",SaveMode.Append,"ds")
43.     sql = s"select id,code,customer_code,name,mobile,type,organization_id,organization_name,create_time,update_time,deleted,status,customer_name," +
44.     s"customer_id,channel_type,nike_name,version,register_Time,'${date}' AS ds from TABLENAME where date(update_time) ='${date}'"
45.     MysqlToHiveTools.readFromMysqlIncrement(spark,dataSource,sql.replace("TABLENAME","t_account"),
46.                                             s"${odsDatabaseName}.ods_t_account",SaveMode.Append,"ds")
47.   }
51.   /**
52.    * 数据etl
53.    * @param spark SparkSession实例
54.    * @param SparkSession 数据库配置
55.    */
56.   def etl(spark: SparkSession, properties:Properties): Unit = {
57.     val sparkConf = spark.sparkContext.getConf
58.     // 获取同步的日期
59.     var lastDate = sparkConf.get("spark.etl.last.day", DateUtils.getLastDayString)
60.     val dateList = new  ListBuffer[String]()
61.     if(lastDate.isEmpty){
62.       // 未配置，设置为前一天
63.       lastDate = DateUtils.getLastDayString
64.     }
65.     if(lastDate.contains("~")){
66.       // 如果是时间段，获取时间段中的每一天，解析为时间list
67.       val dateArray = lastDate.split("~")
68.       DateUtils.findBetweenDates(dateArray(0), dateArray(1)).foreach(it => dateList.append(it))
69.     }else if(lastDate.contains(",")){
70.       // 如果是使用,分隔的多个日期，解析为时间list
71.       lastDate.split(",").foreach(it => dateList.append(it))
72.     }else{
73.       // 添加进时间列表
74.       dateList.append(lastDate)
75.     }
76.     // 循环同步每天的数据
77.     dateList.foreach(it =>  loadData(spark, properties, it))
78.   }
81.   def main(args: Array[String]): Unit = {
82.     job() {
83.       val sparkAndProperties = SparkUtils.get()
84.       val spark = sparkAndProperties.spark
85.       val properties = sparkAndProperties.properties
86.       // 调度任务
87.       etl(spark,properties)
88.     }
89.   }
90. }

复制代码

删除Partition的代码示例：

1. object DropPartitionTools {
4.   /**
5.    * 删除指定的Partition
6.    * @param SparkSession实例
7.    * @param database数据库名称
8.    * @param table表名称
9.    * @param partitionKey 分区字段的名称
10.    * @param partitionValue 具体的分区值
11.    */
12.   def dropPartitionIfExists(spark: SparkSession, database: String, table: String, partitionKey: String, partitionValue:String): Unit ={
14.      val df = spark.sql(
15.        s"""
16.          | show tables in ${database} like '${table}'
17.          |""".stripMargin)
19.     if(df.count() > 0 ){
20.       // 表存在，删除分区
21.       spark.sql(
22.         s"""
23.            |ALTER TABLE  ${database}.${table} DROP  IF EXISTS  PARTITION (${partitionKey}='${partitionValue}')
24.            |""".stripMargin)
25.     }
26.   }
29.   /**
30.    * 删除Partition
31.    * @param SparkSession实例
32.    * @param database数据库名称
33.    * @param table表名称
34.    * @param partitionKey 分区字段的名称
35.    */
36.   def dropHistoryPartitionIfExists(spark: SparkSession, database: String, table: String, partitionKey: String): Unit ={
38.     val df = spark.sql(
39.       s"""
40.          | show tables in ${database} like '${table}'
41.          |""".stripMargin)
43.     if(df.count() > 0 ){
44.       // 表存在，删除历史分区，获取8天前的日期
45.       val sevenDay = DateUtils.getSomeLastDayString(8);
46.       spark.sql(
47.         s"""
48.            |ALTER TABLE  ${database}.${table} DROP  IF EXISTS  PARTITION (${partitionKey} ='${sevenDay}')
49.            |""".stripMargin)
50.     }
51.   }
53. }

复制代码

从RDS同步数据到HIVE的代码示例：

1. object MysqlToHiveTools {
4.   /**
5.    * 从mysql抽取数据到hive -- 全量
6.    * @param spark spark实例
7.    * @param dataSource 数据库配置信息
8.    * @param tableName 抽取的数据库表名
9.    * @param destTableName 目标表名
10.    * @param mode 抽取的模式
11.    */
12.   def mysqlToHiveTotal(spark: SparkSession, dataSource: JSONObject,tableName: String, destTableName:String,mode: SaveMode, partition: String): Unit = {
13.      val sql = "(select * from " + tableName + ") as t"
14.      mysqlToHive(spark, dataSource, sql, destTableName, mode, partition)
15.   }
18.   /**
19.    * 从mysql抽取数据到hive -- 增量量
20.    * @param spark spark实例
21.    * @param dataSource 数据库配置信息
22.    * @param sql 抽取数据的SQL
23.    * @param destTableName 目标表名
24.    * @param mode 抽取的模式
25.    */
26.   def readFromMysqlIncrement(spark: SparkSession, dataSource: JSONObject,sql: String, destTableName:String,mode: SaveMode, partition: String): Unit = {
27.     mysqlToHive(spark, dataSource, sql, destTableName, mode, partition)
28.   }
31.   /**
32.    * 真正的抽取数据
33.    * @param spark spark实例
34.    * @param properties 数据库配置信息
35.    * @param sql 抽取数据的SQL
36.    * @param destTableName 目标表名
37.    * @param mode 抽取的模式
38.    */
39.   def mysqlToHive(spark: SparkSession, dataSource: JSONObject,sql: String, destTableName:String, mode: SaveMode, partition: String):Unit={
40.     val df = spark.read.format("jdbc")
41.       .option("url",dataSource.getString("url"))
42.       .option("driver",dataSource.getString("driver"))
43.       .option("fetchSize", 10000)
44.       .option("numPartitions",2)
45.       .option("dbtable",s"(${sql}) AS t")
46.       .option("user",dataSource.getString("user"))
47.       .option("password",dataSource.getString("password"))
48.       .load()
49.     if(partition == null || partition.isEmpty){
50.       df.write.format("parquet").mode(mode).saveAsTable(destTableName)
51.     }else{
52.       df.write.format("parquet").mode(mode).partitionBy("ds").saveAsTable(destTableName)
53.     }
54.   }
55. }

复制代码

Spark Application代码示例

1. trait SparkHivePartitionOverwriteApplication extends Logging{
4.   def getProperties(): Properties ={
5.     val prop:Properties = new Properties()
6.     val inputStream = this.getClass.getClassLoader.getResourceAsStream("config.properties")
7.     prop.load(inputStream);
8.     prop
9.   }
11.   def job(appName: String = null,
12.           master: String = null)(biz: => Unit): Unit = {
13.     var spark: SparkSession = null
14.     System.setProperty("HADOOP_USER_NAME", "mapred")
15.     val prop:Properties = getProperties()
16.     if (null == appName) {
17.       spark = SparkSession.builder
18.         .config("spark.sql.parquet.writeLegacyFormat", true)
19.         .config("spark.sql.sources.partitionOverwriteMode","dynamic")
20.         .config("hive.exec.dynamic.partition.mode","nonstrict")
21.         .config("spark.sql.hive.convertMetastoreParquet",false)
22.         .enableHiveSupport
23.         .getOrCreate
24.       var sparkAndProperties = SparkAndProperties(spark, prop)
25.       SparkUtils.set(sparkAndProperties)
26.     } else {
27.       spark = SparkSession.builder.master(master).appName(appName)
28.         .config("spark.sql.parquet.writeLegacyFormat", true)
29.         .config("spark.sql.sources.partitionOverwriteMode","dynamic")
30.         .config("hive.exec.dynamic.partition.mode","nonstrict")
31.         .config("spark.sql.hive.convertMetastoreParquet",false)
32.         .config("spark.testing.memory","2147480000")
33.         .config("spark.driver.memory","2147480000")
34.         .enableHiveSupport.getOrCreate
35.       var sparkAndProperties = SparkAndProperties(spark, prop)
36.       SparkUtils.set(sparkAndProperties)
37.       SparkUtils.set(sparkAndProperties)
38.     }
39.     biz
40.     spark.stop()
41.     SparkUtils.remove()
42.   }
44. }
46. case class SparkAndProperties(spark: SparkSession,
47.                               properties: Properties)

复制代码

四、配套生态

• 自定义UDF函数
  

在使用的过程中，需要将表中的IP地址，解析为地点地的名称，这需要调用第三方的一个服务接口来完成，为了完成这个任务，定义了一个自定义UDF函数，举行解析。
a. 自定义UDF函数

1. object ParseIp  {
2.     def evaluate(ip: String):String= {
3.       // 具体的IP解析服务
4.       SplitAddress.getPlaceFromIp(ip)
5.    }
6. }

复制代码

b. 使用自定义UDF函数

1. object TraceTmpEtl extends SparkHivePartitionOverwriteApplication{
3.   /**
4.    * 数据同步任务
5.    * @param spark sparkSession实例
6.    * @param properties 数据库配置
7.    * @param date 日期
8.    */
9.   def tmp_t_trace_user_visit_real_time_statistic(spark: SparkSession,properties:Properties,date: String):Unit ={
10.     // 获取数据库配置的数据库名称
11.     val odsDatabaseName = properties.get("hive.datasource.ods")
12.     val tmpDatabaseName = properties.get("hive.datasource.tmp")
14.     // 注册自定义的UDF函数
15.     spark.udf.register("parseIP", (ip: String) => SplitAddress.getPlaceFromIp(ip))
16.     // 在Spark SQL中使用UDF函数
17.     spark.sql(
18.       s"""
19.          |INSERT OVERWRITE TABLE ${tmpDatabaseName}.tmp_t_statistic partition(ds='${date}')
20.          |select
21.          |          `id` ,
22.          |          `create_time` ,
23.          |          `update_time` ,
24.          |          `ip` ,
25.          |      replace( replace( replace(replace( case when parseIP(ip) rlike '^中国' then replace(parseIP(ip),'中国','')
26.          |          when parseIP(ip) rlike '^内蒙古' then replace(parseIP(ip),'内蒙古','内蒙古自治区')
27.          |          when parseIP(ip) rlike '^广西' then replace(parseIP(ip),'广西','广西壮族自治区')
28.          |          when parseIP(ip) rlike '^西藏' then replace(parseIP(ip),'西藏','西藏自治区')
29.          |          when parseIP(ip) rlike '^宁夏' then replace(parseIP(ip),'宁夏','宁夏回族自治区')
30.          |          when parseIP(ip) rlike '^新疆' then replace(parseIP(ip),'新疆','新疆维吾尔自治区')
31.          |          when parseIP(ip) rlike '^香港' then replace(parseIP(ip),'香港','香港特别行政区')
32.          |          when parseIP(ip) rlike '^澳门' then replace(parseIP(ip),'澳门','澳门特别行政区')
33.          |     else parseIP(ip) end, "省", "省."),"市", "市."),"县", "县."),"区", "区.") as ip_place,
34.          |          `page_view`
35.          |from ${odsDatabaseName}.ods_t_statistic where ds ='${date}'
36.          |""".stripMargin)
37.   }
39.   /**
40.    * 数据etl
41.    * @param spark SparkSession实例
42.    * @param properties 数据库配置
43.    */
44.   def etl(spark: SparkSession, properties:Properties): Unit = {
45.     val lastDate = DateUtils.getLastDayString
46.     tmp_t_trace_user_visit_real_time_statistic(spark,properties, lastDate)
47.   }
50.   
51.   def main(args: Array[String]): Unit = {
52.     job() {
53.       val sparkAndProperties = SparkUtils.get()
54.       val spark = sparkAndProperties.spark
55.       val properties = sparkAndProperties.properties
56.       etl(spark,properties)
57.     }
58.   }
59. }

复制代码

• 数据库的设置安全性问题
  

刚开始数据库设置同步设置文件直接写死，但是后续发现这样存在一些安全性的问题，后来采用将数据库相关的设置组合为一个JSON字符串，将其加密之后生存到MongoDB中，在使用时举行查询解密。

1. public class DataSourceUtils {
3.     private  static Logger logger = LoggerFactory.getLogger(DataSourceUtils.class);
5.     public static JSONObject getDataSourceProperties(String dataSourceKey,Properties properties){
6.         List<ServerAddress> adds = new ArrayList<>();
7.         try {
8.             String filePath = properties.getProperty("spark.mongo.properties.file.url");
9.             properties = new Properties();
10.             File file = new File(filePath);
11.             FileInputStream inputStream = null;
12.              inputStream = new FileInputStream(file);
13.             properties.load(inputStream);
14.         }catch (Exception e){
15.             logger.info("not load file, reason：" + e.getMessage());
16.             e.printStackTrace();
17.         }
18.         String mongoUrl = properties.getProperty("mongo_url");
19.         String mongoPort = properties.getProperty("mongo_port");
20.         String mongoDbName = properties.getProperty("mongo_dbName");
21.         String mongoCollect = properties.getProperty("mongo_collect");
22.         String mongoUser = properties.getProperty("mongo_user");
23.         String mongoPassword = properties.getProperty("mongo_password");
24.         String desKey = properties.getProperty("data_des_key");
25.         ServerAddress serverAddress = new ServerAddress(mongoUrl, Integer.parseInt(mongoPort));
26.         adds.add(serverAddress);
27.         List<MongoCredential> credentials = new ArrayList<>();
28.         MongoCredential mongoCredential = MongoCredential.createScramSha1Credential(mongoUser, mongoDbName, mongoPassword.toCharArray());
29.         credentials.add(mongoCredential);
30.         MongoClient mongoClient = new MongoClient(adds, credentials);
31.         MongoDatabase mongoDatabase = mongoClient.getDatabase(mongoDbName);
32.         MongoCollection<Document> collection = mongoDatabase.getCollection(mongoCollect);
33.         //指定查询过滤器
34.         Bson filter = Filters.eq("key", dataSourceKey);
35.         //指定查询过滤器查询
36.         FindIterable findIterable = collection.find(filter);
37.         //取出查询到的第一个文档
38.         Document document = (Document) findIterable.first();
39.         //打印输出
40.         String content = DESUtil.decrypt(desKey, document.getString("content"));
41.         return JSON.parseObject(content);
42.     }
45.     public static  Properties json2Properties(JSONObject jsonObject){
46.         String tmpKey = "";
47.         String tmpKeyPre = "";
48.         Properties properties = new Properties();
49.         j2p(jsonObject, tmpKey, tmpKeyPre, properties);
50.         return properties;
51.     }
55.     private static void j2p(JSONObject jsonObject, String tmpKey, String tmpKeyPre, Properties properties){
56.         for (String key : jsonObject.keySet()) {
57.             // 获得key
58.             String value = jsonObject.getString(key);
59.             try {
60.                 JSONObject jsonStr = JSONObject.parseObject(value);
61.                 tmpKeyPre = tmpKey;
62.                 tmpKey += key + ".";
63.                 j2p(jsonStr, tmpKey, tmpKeyPre, properties);
64.                 tmpKey = tmpKeyPre;
65.             } catch (Exception e) {
66.                 properties.put(tmpKey + key, value);
67.                 System.out.println(tmpKey + key + "=" + value);
68.             }
69.         }
70.     }
71.     public static void main(String[] args) {
73.     }
74. }

复制代码

• Spark任务脚本示例
  

1. #!/bin/sh
3. ##### env ###########
4. export JAVA_HOME=/usr/java/jdk1.8.0_151
5. export SPARK_HOME=/opt/cloudera/parcels/CDH/lib/spark
6. export PATH=${JAVA_HOME}/bin:${SPARK_HOME}/bin:${PATH}
7. export SPARK_USER=hadoop
8. export HADOOP_USER_NAME=hadoop
9. LAST_DAY="$1"
10. echo LAST_DAY
12. spark-submit \
13. --class net.app315.bigdata.operatereport.ods.MarketMysqlToHiveEtl \
14. --conf spark.sql.hive.metastore.version=2.1.1 \
15. --conf spark.sql.hive.metastore.jars=/opt/cloudera/parcels/CDH/lib/hive/lib/* \
16. --jars /opt/cloudera/parcels/CDH/lib/spark/jars/mysql-connector-java-5.1.48.jar,/opt/cloudera/parcels/CDH/lib/spark/jars/druid-1.1.10.jar \
17. --master yarn \
18. --deploy-mode cluster \
19. --executor-memory 4G \
20. --driver-memory 2G \
21. --num-executors 4 \
22. --executor-cores 2 \
23. --conf spark.dynamicAllocation.minExecutors=1 \
24. --conf spark.dynamicAllocation.maxExecutors=8 \
25. --conf spark.yarn.am.attemptFailuresValidityInterval=1h \
26. --conf spark.yarn.max.executor.failures=128 \
27. --conf spark.yarn.executor.failuresValidityInterval=1h \
28. --conf spark.task.maxFailures=4 \
29. --conf spark.yarn.maxAppAttempts=2 \
30. --conf spark.scheduler.mode=FIFO \
31. --conf spark.network.timeout=420000 \
32. --conf spark.dynamicAllocation.enabled=true \
33. --conf spark.executor.heartbeatInterval=360000 \
34. --conf spark.sql.crossJoin.enabled=true \
35. --conf spark.mongo.properties.file.url=/opt/conf/mongo.properties \
36. --conf spark.etl.last.day="${LAST_DAY}" \
37. ./target/spark-operate-report-project-1.0.jar

复制代码

• Job任务脚本实例
  

1. nodes:
3.   - name: bigdata_market_ods_etl
4.     type: command
5.     config:
6.       command: sh -x ./script/bigdata_market_ods_etl.sh "${spark.etl.last.day}"
7.       failure.emails: mxx@xxx.com
9.   - name: bigdata_market_dim_etl
10.     type: command
11.     config:
12.       command: sh -x ./script/bigdata_market_dim_etl.sh "${spark.etl.last.day}"
13.       failure.emails: mxx@xxx.com
14.     dependsOn:
15.           - bigdata_market_ods_etl
16.          
17.   - name: bigdata_market_dw_etl
18.     type: command
19.     config:
20.       command: sh -x ./script/bigdata_market_dw_etl.sh "${spark.etl.last.day}"
21.       failure.emails: mxx@xxx.com
22.     dependsOn:
23.           - bigdata_market_dim_etl
24.           - bigdata_user_dw_etl

复制代码

五、备注

• Davinci报表   一个开源的报表平台
  

第二代-基于DolphinScheduler的离线数据同步

一、背景

自从上次开始使用基于Hadoop的大数据体现方案之后，业务安稳发展，但是随着时间的推移，新的问题开始出现，重要出现的问题为两个：

• 数据的变更越来越频繁，基于之前SparkSQL任务的方式，只要需要对表结构举行变更，就需要重新修改Scala代码，然后重新举行任务的打包，这对于一些不认识代码的人来说，不太友好，而且本钱也很高。
  
• 虽然使用了Presto对HIVE的数据查询举行了加速，但是地点数据量越来越大，分析要求越来越复杂，即席查询越来越多，由于集群本身资源有限，查询能力出现了显著瓶颈。
  

二、数据同步架构

随着技术的发展已经对大数据的认识，接触到了更多的大数据相关的知识与组件，基于此，通过认真分析与思考之后，对数据的同步方案举行了如下的重新设计。

• 数据存储与查询放弃了HDFS+HIVE+Presto的组合，转而采用现代化的MPP数据库StarRocks，StarRocks在数据查询的效率层面非常良好，在相同资源的情况下，可以解决现在遇到的数据查询瓶颈。
  
• 数据同步放弃了SparkSQL，转而采用更加轻量级的DATAX来举行，其只需要通过简朴的设置，即可完成数据的同步，同时其也支持StarRocks Writer，开辟职员只需要具备简朴的SQL知识，就可以完成整个数据同步任务的设置，难度大大降低，效率大大提升，友好度大大提升。
  
• 定时任务调度放弃Azkaban，采用现代化的任务调度工作Apache DolphinScheduler，通过可视化的页面举行调度任务工作流的设置，更加友好。
  

三、数据同步的具体流程

数据同步在这种方式下变动非常简朴，只需要可视化的设置DataX任务，即可自动调度。下面的一个任务的设置示例

1. {
2.   "job": {
3.     "setting": {
4.       "speed": {
5.         "channel":1
6.       }
7.     },
8.     "content": [
9.       {
10.         "reader": {
11.           "name": "mysqlreader",
12.           "parameter": {
13.             "username": "",
14.             "password": "",
15.             "connection": [
16.               {
17.                 "querySql": [
18.                   "SELECT CustomerId AS customer_id FROM base_info.base_customer where date(UpdateTime) > '${sdt}' and date(UpdateTime) < '${edt}'"
19.                 ],
20.                 "jdbcUrl": [
21.                   "jdbc:mysql://IP:3306/base_info?characterEncoding=utf-8&useSSL=false&tinyInt1isBit=false"
22.                 ]
23.               }
24.             ]
25.           }
26.         },
27.         "writer": {
28.           "name": "starrockswriter",
29.           "parameter": {
30.             "username": "xxx",
31.             "password": "xxx",
32.             "database": "ods_cjm_test",
33.             "table": "ods_base_customer",
34.             "column": ["id"],
35.             "preSql": [],
36.             "postSql": [],
37.             "jdbcUrl": "jdbc:mysql://IP:9030/",
38.             "loadUrl": ["IP:8050", "IP:8050", "IP:8050"],
39.             "loadProps": {
40.               "format": "json",
41.               "strip_outer_array": true
42.             }
43.           }
44.         }
45.       }
46.         ]
47.     }
48. }

复制代码

数据同步过程中，遇到了另外一个问题，即业务存在大量的分库分表的，这些分库分表的逻辑五花八门，60张左右的逻辑板，经太过库分表之后达到了惊人的5000多张，为每张表设置任务很显然不太正常，这就需要能够在举行数据同步的时候动态生成需要的表列表，把表列表设置到DataX的设置文件中去。
颠末技术的调用，Apache DolphinScheduler的Python任务类型很适合做这个事情，由于公司本身使用了Apache DolphinScheduler3.0的版本，其Python任务还不支持返回数据到鄙俚节点，但是社区最新版本已经支持该能力，因为按照已实现版本对其举行改造。
改造之后，Python节点能够将数据通报给他的鄙俚节点，因此使用Python脚本查询获取需要举行同步的表列表，将其通报给DataX节点，完成动态表的数据同步

1. import pymysql
2. import datetime
5. def select_all_table(date: str):
6.     result_list = []
7.     sql = """
8.     SELECT concat('"', table_name, '"')
9.     FROM information_schema.`TABLES`
10.     WHERE table_schema='hydra_production_flow'
11.         and table_name like 't_package_flow_log_%'
12.         and table_name like '%_{}'
13.     """.format(date)
14.     conn = pymysql.connect(host='', port=3306, user='', passwd='',
15.                            db='information_schema')
16.     cur = conn.cursor()
17.     cur.execute(query=sql)
18.     while 1:
19.         res = cur.fetchone()
20.         if res is None:
21.             break
22.         result_list.append(res[0])
23.     cur.close()
24.     conn.close()
25.     return result_list
28. if __name__ == '__main__':
29.     # 获取当前年月
30.     # 获取当前日期
31.     today = datetime.date.today()
32.     # 计算前一天的日期
33.     yesterday = today - datetime.timedelta(days=1)
34.     current_date = yesterday.strftime("%Y_%m")
35.     table_list = select_all_table(current_date)
36.     table_str = ",".join(table_list)
37.     # 设置变量,传递给下游节点
38.     print('${setValue(table_list=%s)}' % table_str)

复制代码

1. {
2.   "job": {
3.     "setting": {
4.       "speed": {
5.         "channel":1
6.       }
7.     },
8.     "content": [
9.       {
10.         "reader": {
11.           "name": "mysqlreader",
12.           "parameter": {
13.             "username": "xxx",
14.             "password": "xxxx",
15.             "column": [
16.               "id",
17.               "concat('t_package_flow_log_',DATE_FORMAT(create_time,'%Y_%m'))",
18.               "operation_type"
19.             ],
20.             "where": "date(create_time) ${operator_symbol} '${dt}'",
21.             "connection": [
22.               {
23.                 "table": [
24.                   ${table_list}
25.                 ],
26.                 "jdbcUrl": [
27.                   "jdbc:mysql://xx:3306/hydra_production_flow?characterEncoding=utf-8&useSSL=false&tinyInt1isBit=false"
28.                 ]
29.               }
30.             ]
31.           }
32.         },
33.         "writer": {
34.                     "name": "starrockswriter",
35.                     "parameter": {
36.                         "username": "xxxxxx",
37.                         "password": "xxxxxxx",
38.                         "database": "ods_cjm",
39.                         "table": "ods_t_package_flow_log",
40.                         "column": ["id", "table_name","operation_type"],
41.                         "preSql": [],
42.                         "postSql": [],
43.                         "jdbcUrl": "jdbc:mysql://IP:9030/",
44.                         "loadUrl": ["IP:8050", "IP:8050", "IP:8050"],
45.                         "loadProps": {
46.                             "format": "json",
47.                             "strip_outer_array": true
48.                         }
49.                     }
50.                 }
51.             }
52.         ]
53.     }
54. }

复制代码

四、踩坑记录

• DATAX只支持python2.x
  

下载支持python3.x的相关文件，替换DataX中的相同文件，即可支持python3.x使用
五、备注

• StarRocks    高性能的MPP数据库
  
• DataX  离线数据同步
  
• Apache DolphinScheduler  任务调度工具
  

第三代-基于Python自定义的离线数据同步

一、背景

自从采用Apache DolphinScheduler + StarRocks数据方案以来，统统都很安稳发展；但是随着时间的推移，总会出现新的问题。
随着数据量的增多，使用方需求的增长，已经一些其他因素的影响，对现在的数据同步架构带来了一些不小的挑战，这些问题导致任务的维护和更新越来越贫苦，需要耗费大量的时间来举行，急需一种新的方式来处理。

• 由于等保的要求，线上RDS数据库不再支持通过公网访问，又因为StarRocks也在内网，这就导致了之前的数据同步链路彻底断裂，需要新的方案。
  
• 由于数据结构的频繁变更、服务器资源导致的任务调度非常等等原因，需要重跑数据的需求越来越多，这就导致需要不停的修改任务的调度参数(如日期)，现在已经上线了10个业务的调度任务，也就是重新同步一次，就需要依次修改调度这10个任务，这期间还需要专人举行状态的跟踪，纵然修改调度，压力很大。
  

二、数据同步架构

鉴于数据链路变更，导致原本数据链路断裂的问题，通过调研之后，决定采用KAFKA举行数据的中转，在内网部署KAFKA集群，同时该集群提供公网访问地址；在RDS地点的内网呆板上使用DataX将RDS数据通过公网地址写入KAFKA，在内网中通过KafkaConnector消耗数据写入StarRocks。
鉴于新的资源有限，原本内网提供了4台8C32G的服务器，但是新的RDS地点内网只能提供一台最大4C8G的服务器。因此放弃了使用Apache DolphinScheduler来举行调度，直接使用crontab调用对应的Python脚本举行DataX任务调度。

三、具体的数据同步

新的方案，重要解决的问题有两个，一是DataX怎样将数据写入KAFKA，二是Python脚本怎么解决前面遇到的修改复杂的问题。

• DataX写KAFKA
  

DataX本身并没有kafkawriter实现，这就需要我们本身实现一个KafkaWriter来支持我们的需求，同时为了数据安全，希望能够对数据举行加密。
DataX的KafkaWriter实现

1. public class KafkaWriter extends Writer {
3.     public static class Job extends Writer.Job {
5.         private static final Logger logger = LoggerFactory.getLogger(Job.class);
6.         private Configuration conf = null;
8.         @Override
9.         public List<Configuration> split(int mandatoryNumber) {
10.             List<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber);
11.             for (int i = 0; i < mandatoryNumber; i++) {
12.                 configurations.add(conf);
13.             }
14.             return configurations;
15.         }
17.         private void validateParameter() {
18.             this.conf.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaWriterErrorCode.REQUIRED_VALUE);
19.             this.conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
20.         }
22.         @Override
23.         public void init() {
24.             this.conf = super.getPluginJobConf();
25.             logger.info("kafka writer params:{}", conf.toJSON());
26.             this.validateParameter();
27.         }
30.         @Override
31.         public void destroy() {
33.         }
34.     }
36.     public static class Task extends Writer.Task {
37.         private static final Logger logger = LoggerFactory.getLogger(Task.class);
38.         private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");
40.         private Producer<String, String> producer;
41.         private String fieldDelimiter;
42.         private Configuration conf;
43.         private Properties props;
44.         private AesEncryption aesEncryption;
45.         private List<String> columns;
47.         @Override
48.         public void init() {
49.             this.conf = super.getPluginJobConf();
50.             fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
51.             columns = conf.getList(Key.COLUMN_LIST, new ArrayList<>(), String.class);
53.             props = new Properties();
54.             props.put("bootstrap.servers", conf.getString(Key.BOOTSTRAP_SERVERS));
55.             props.put("acks", conf.getUnnecessaryValue(Key.ACK, "0", null));//这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
56.             props.put("retries", conf.getUnnecessaryValue(Key.RETRIES, "5", null));
57.             props.put("retry.backoff.ms", "1000");
58.             props.put("batch.size", conf.getUnnecessaryValue(Key.BATCH_SIZE, "16384", null));
59.             props.put("linger.ms", 100);
60.             props.put("connections.max.idle.ms", 300000);
61.             props.put("max.in.flight.requests.per.connection", 5);
62.             props.put("socket.keepalive.enable", true);
63.             props.put("key.serializer", conf.getUnnecessaryValue(Key.KEYSERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
64.             props.put("value.serializer", conf.getUnnecessaryValue(Key.VALUESERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
65.             producer = new KafkaProducer<String, String>(props);
66.             String encryptKey = conf.getUnnecessaryValue(Key.ENCRYPT_KEY, null, null);
67.             if(encryptKey != null){
68.                 aesEncryption = new AesEncryption(encryptKey);
69.             }
70.         }
72.         @Override
73.         public void prepare() {
74.             AdminClient adminClient = AdminClient.create(props);
75.             ListTopicsResult topicsResult = adminClient.listTopics();
76.             String topic = conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
77.             try {
78.                 if (!topicsResult.names().get().contains(topic)) {
79.                     new NewTopic(
80.                             topic,
81.                             Integer.parseInt(conf.getUnnecessaryValue(Key.TOPIC_NUM_PARTITION, "1", null)),
82.                             Short.parseShort(conf.getUnnecessaryValue(Key.TOPIC_REPLICATION_FACTOR, "1", null))
83.                     );
84.                     List<NewTopic> newTopics = new ArrayList<NewTopic>();
85.                     adminClient.createTopics(newTopics);
86.                 }
87.                 adminClient.close();
88.             } catch (Exception e) {
89.                 throw new DataXException(KafkaWriterErrorCode.CREATE_TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE.getDescription());
90.             }
91.         }
93.         @Override
94.         public void startWrite(RecordReceiver lineReceiver) {
95.             logger.info("start to writer kafka");
96.             Record record = null;
97.             while ((record = lineReceiver.getFromReader()) != null) {
98.                 if (conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null)
99.                         .equalsIgnoreCase(WriteType.TEXT.name())) {
100.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
101.                             Md5Encrypt.md5Hexdigest(recordToString(record)),
102.                             aesEncryption ==null ? recordToString(record): JSONObject.toJSONString(aesEncryption.encrypt(recordToString(record))))
103.                     );
104.                 } else if (conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null)
105.                         .equalsIgnoreCase(WriteType.JSON.name())) {
106.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
107.                             Md5Encrypt.md5Hexdigest(recordToString(record)),
108.                             aesEncryption ==null ? recordToJsonString(record) : JSONObject.toJSONString(aesEncryption.encrypt(recordToJsonString(record))))
109.                     );
110.                 }
111.                 producer.flush();
112.             }
113.         }
115.         @Override
116.         public void destroy() {
117.             if (producer != null) {
118.                 producer.close();
119.             }
120.         }
122.         /**
123.          * 数据格式化
124.          *
125.          * @param record
126.          * @return
127.          */
128.         private String recordToString(Record record) {
129.             int recordLength = record.getColumnNumber();
130.             if (0 == recordLength) {
131.                 return NEWLINE_FLAG;
132.             }
133.             Column column;
134.             StringBuilder sb = new StringBuilder();
135.             for (int i = 0; i < recordLength; i++) {
136.                 column = record.getColumn(i);
137.                 sb.append(column.asString()).append(fieldDelimiter);
138.             }
140.             sb.setLength(sb.length() - 1);
141.             sb.append(NEWLINE_FLAG);
142.             return sb.toString();
143.         }
145.         /**
146.          * 数据格式化
147.          *
148.          * @param record 数据
149.          *
150.          */
151.         private String recordToJsonString(Record record) {
152.             int recordLength = record.getColumnNumber();
153.             if (0 == recordLength) {
154.                 return "{}";
155.             }
156.             Map<String, Object> map = new HashMap<>();
157.             for (int i = 0; i < recordLength; i++) {
158.                 String key = columns.get(i);
159.                 Column column = record.getColumn(i);
160.                 map.put(key, column.getRawData());
161.             }
162.             return JSONObject.toJSONString(map);
163.         }
164.     }
165. }

复制代码

举行数据加密的实现：

1. public class AesEncryption {
3.     private SecretKey secretKey;
5.     public AesEncryption(String secretKey) {
6.         byte[] keyBytes = Base64.getDecoder().decode(secretKey);
7.         this.secretKey = new SecretKeySpec(keyBytes, 0, keyBytes.length, "AES");
8.     }
11.     public String encrypt(String data) {
12.         try {
13.             Cipher cipher = Cipher.getInstance("AES");
14.             cipher.init(Cipher.ENCRYPT_MODE, secretKey);
15.             byte[] encryptedBytes = cipher.doFinal(data.getBytes());
16.             return Base64.getEncoder().encodeToString(encryptedBytes);
17.         } catch (Exception e) {
18.             throw new RuntimeException(e);
19.         }
20.     }
22.     public String decrypt(String encryptedData) throws Exception {
23.         Cipher cipher = Cipher.getInstance("AES");
24.         cipher.init(Cipher.DECRYPT_MODE, secretKey);
25.         byte[] decodedBytes = Base64.getDecoder().decode(encryptedData);
26.         byte[] decryptedBytes = cipher.doFinal(decodedBytes);
27.         return new String(decryptedBytes);
28.     }
29. }

复制代码

Kafka的公网设置
Kafka的表里网设置，只需要修改kafka/config下面的server.properties文件中的如下设置即可。

1. # 配置kafka的监听端口,同时监听9093和9092
2. listeners=INTERNAL://kafka节点3内网IP:9093,EXTERNAL://kafka节点3内网IP:9092
4. # 配置kafka的对外广播地址， 同时配置内网的9093和外网的19092
5. advertised.listeners=INTERNAL://kafka节点3内网IP:9093,EXTERNAL://公网IP:19092
7. # 配置地址协议
8. listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
10. # 指定broker内部通信的地址
11. inter.broker.listener.name=INTERNAL

复制代码

• 自定义的设置文件
  

Python脚本需要能够自动生成对应的DataX调度的设置文件和shell脚本，自动调度DataX举行任务的执行。因此颠末调研，采用自定义设置文件，通过读取设置文件，动态生成对应的DataX任务脚本和调度脚本，调度任务执行。
自定义的设置文件示例1：

1. {
2.   "datasource": {
3.     "host": "xxxxxx",
4.     "port": "3306",
5.     "username": "xxxxx",
6.     "password": "xxxxxxx",
7.     "properties": {
8.       "characterEncoding": "utf-8",
9.       "useSSL": "false",
10.       "tinyInt1isBit": "false"
11.     }
12.   },
13.   "table": {
14.     "database": "app",
15.     "table": "device",
16.     "column": [
17.       "Id AS id",
18.       "CompanyName AS company_name",
19.       "CompanyId AS company_id",
20.       "SecretKey AS secret_key",
21.       "Brand AS brand",
22.       "ModelType AS model_type",
23.       "Enable AS enable",
24.       "CAST(CreateTime as CHAR) AS create_time",
25.       "CAST(UpdateTime as CHAR) AS update_time"
26.     ],
27.     "where": "date(UpdateTime) >= '$[yyyy-MM-dd-8]'",
28.     "searchTableSql": []
29.   },
30.   "kafka": {
31.     "topic": "mzt_ods_cjm.ods_device"
32.   }
33. }

复制代码

支持分库分表的设置文件示例2

1. {
2.   "datasource": {
3.     "host": "xxxxxxx",
4.     "port": "3306",
5.     "username": "xxxxxxx",
6.     "password": "xxxxxxxx",
7.     "properties": {
8.       "characterEncoding": "utf-8",
9.       "useSSL": "false",
10.       "tinyInt1isBit": "false"
11.     }
12.   },
13.   "table": {
14.     "database": "hydra_logistics_flow",
15.     "table": "",
16.     "column": [
17.       "id",
18.       "concat('t_logistics_sweep_out_code_flow_',DATE_FORMAT(create_time,'%Y')) AS table_name",
19.       "cus_org_id",
20.       "CAST(create_time as CHAR) AS create_time",
21.       "replace_product_id",
22.       "replace_product_name",
23.       "replace_product_code"
24.     ],
25.     "where": "date(create_time) >= '$[yyyy-MM-dd-8]'",
26.     "searchTableSql": [
27.       "SELECT concat('t_logistics_sweep_out_code_flow_',YEAR(SUBDATE(CURDATE(), 1))) AS TABLE_NAME",
28.       "SELECT concat('t_logistics_sweep_out_code_flow_',YEAR(DATE_SUB(DATE_SUB(CURDATE(), INTERVAL 1 DAY), INTERVAL 1 YEAR))) AS TABLE_NAME"
29.     ]
30.   },
31.   "kafka": {
32.     "topic": "mzt_ods_cjm.ods_t_logistics_sweep_out_code_flow"
33.   }
34. }

复制代码

如上的设置文件，表明如下：
KEY说明datasourceRDS数据源datasource.hostRDS数据库的hostdatasource.port>RDS数据库的端口datasource.usernameRDS数据库的用户名datasource.passwordRDS数据库的密码datasource.propertiesjdbc连接的参数，连接时拼接为?key=value&key=valuetable要同步的表信息table.databaseRDS数据库名称table.tableRDS中表的名称，分库分表的可以为空table.columnRDS表中要同步的字段列表，支持取别名和使用函数table.where同步数据的过滤条件table.searchTableSql查询表名称的SQL语句，用于动态分库分表kafkakafka相关的设置kafka.topic数据要写入的kafka topic的名称

• Python调度脚本
  

1. import json
2. import os
3. import pymysql
4. import re
5. from datetime import datetime
6. from dateutil.relativedelta import relativedelta
7. import uuid
8. import subprocess
9. import logging
10. import hmac
11. import hashlib
12. import base64
13. import urllib.parse
14. import urllib
15. import requests
16. import time
17. from typing import List, Mapping
20. def list_files_in_directory(directory_path: str) -> List[str]:
21.     """
22.     获取目录下的所有以.json结尾的文件
23.     :param directory_path: 目录
24.     :return: 文件列表
25.     """
26.     entries = os.listdir(directory_path)
27.     # 过滤出所有文件
28.     files = [entry for entry in entries if
29.              os.path.isfile(os.path.join(directory_path, entry)) and entry.endswith(".json")]
30.     logging.info(f"读取配置文件数量：{len(files)}")
31.     return files
34. def read_file_content(file_path: str) -> str:
35.     """
36.     读取文件内容
37.     :param file_path: 文件路径
38.     :return: 文件内容
39.     """
40.     with open(file_path, 'r', encoding='utf-8') as file:
41.         content = file.read()
42.     return content
45. def read_all_files_in_directory(directory_path: str) -> Mapping[str, str]:
46.     """
47.     读取文件夹下面的所有文件的内容
48.     :param directory_path: 文件夹路径
49.     :return: 内容map
50.     """
51.     logging.info(f"开始读取所有的配置文件信息")
52.     files = list_files_in_directory(directory_path)
53.     file_contents = {}
54.     for file in files:
55.         file_path = os.path.join(directory_path, file)
56.         content = read_file_content(file_path)
57.         file_contents[file] = content
58.     sorted_items = sorted(file_contents.items())
59.     sorted_dict = dict(sorted_items)
60.     return file_contents
63. def search_table_list(datasource: json, search_table_sql_list: List[str]) -> List[str]:
64.     """
65.     执行语句获取表信息
66.     :param datasource: 数据源信息
67.     :param search_table_sql_list: 查询表的SQL语句
68.     :return: 表列表
69.     """
70.     logging.info(f"开始查询需要同步的表")
71.     host = datasource['host']
72.     port = int(datasource['port'])
73.     username = datasource['username']
74.     password = datasource['password']
75.     conn = pymysql.connect(host=host,
76.                            port=port,
77.                            user=username,
78.                            passwd=password,
79.                            db='',
80.                            charset='utf8',
81.                            connect_timeout=200,
82.                            autocommit=True,
83.                            read_timeout=2000
84.                           )
85.     table_name_list = []
86.     for search_table_sql in search_table_sql_list:
87.         search_table_sql = parse_where_sql(search_table_sql)
88.         with conn.cursor() as cursor:
89.             cursor.execute(query=search_table_sql)
90.             while 1:
91.                 res = cursor.fetchone()
92.                 if res is None:
93.                     break
94.                 table_name_list.append(res[0])
95.     return table_name_list
98. def general_default_job_config() -> json:
99.     """
100.     生成默认的datax配置
101.     :return: 默认的配置
102.     """
103.     default_job_json = """
104.     {
105.     "job": {
106.         "setting": {
107.             "speed": {
108.                  "channel":1
109.             }
110.         },
111.         "content": [
112.             {
113.                 "reader": {
114.                     "name": "mysqlreader",
115.                     "parameter": {
116.                         "username": "test",
117.                         "password": "test1234",
118.                         "connection": [
119.                             {
120.                                 "querySql": [
121.                                     "SELECT id, code from test.t_open_api_classify"
122.                                 ],
123.                                 "jdbcUrl": [
124.                                     "jdbc:mysql://IP:3306/test?characterEncoding=utf-8&useSSL=false&tinyInt1isBit=false"
125.                                 ]
126.                             }
127.                         ]
128.                     }
129.                 },
130.                  "writer": {
131.                     "name": "kafkawriter",
132.                     "parameter": {
133.                         "bootstrapServers": "IP:9092,IP:9092,IP:9092",
134.                         "topic": "test-m-t-k",
135.                         "ack": "all",
136.                         "batchSize": 1000,
137.                         "retries": 0,
138.                         "keySerializer": "org.apache.kafka.common.serialization.StringSerializer",
139.                         "valueSerializer": "org.apache.kafka.common.serialization.StringSerializer",
140.                         "fieldDelimiter": ",",
141.                         "writeType": "json",
142.                         "topicNumPartition": 1,
143.                         "topicReplicationFactor": 1,
144.                         "encryptionKey": "5s8FGjerddfWkG/b64CGHHZYvQ=="
145.                     }
146.                 }
147.             }
148.         ]
149.     }
150. }
151.     """
152.     return json.loads(default_job_json, encoding='utf-8')
155. def general_jdbc_url(json_config: json) -> str:
156.     """
157.     根据数据源信息生成jdbc url
158.     :param json_config: 配置
159.     :return: jdbc url
160.     """
161.     logging.info(f"开始解析jdbc url")
162.     host = json_config['datasource']['host']
163.     port = int(json_config['datasource']['port'])
164.     database = json_config['table']['database']
165.     url = "jdbc:mysql://{}:{}/{}".format(host, port, database)
166.     # 解下properties
167.     properties = json_config['datasource']['properties']
168.     properties_list = []
169.     if properties is not None and len(properties) > 0:
170.         for key, value in properties.items():
171.             properties_list.append(key + "=" + str(value))
172.         url = url + "?" + "&".join(properties_list)
173.     logging.info(f"jdbc url： {url}")
174.     return url
177. def parse_where_sql(where_sql: str) -> str:
178.     """
179.     解析where语句
180.     :param where_sql: 原始where语句
181.     :return: 转换之后的where语句
182.     """
183.     # 定义支持的类型 $[yyyyMMdd+N_Y]  $[yyyyMMdd-N_Y]
184.     # 正则表达式模式
185.     logging.info(f"还是解析where语句：where_sql: {where_sql}")
186.     pattern = r"\$\[.*?\]"
187.     return re.sub(pattern, replacement_function, where_sql)
190. def replacement_function(match):
191.     """
192.     替换函数
193.     :param match: 匹配结果
194.     :return: 替换之后的结果
195.     """
196.     matched_text = match.group(0)
197.     return calc_datetime(matched_text)
200. def calc_datetime(expression: str) -> str:
201.     """
202.     计算时间表达式
203.     :param expression: 表达式
204.     :return: 计算之后的值
205.     """
206.     logging.info(f"开始计算时间参数：expression: {expression}")
207.     # 设置映射
208.     format_units = {
209.         "yyyy": "%Y",
210.         "MM": "%m",
211.         "dd": "%d",
212.         "HH": "%H",
213.         "mm": "%M",
214.         "ss": "%S"
215.     }
217.     unit_map = {
218.         "Y": "yyyy",
219.         "M": "MM",
220.         "d": "dd",
221.         "H": "HH",
222.         "m": "mm",
223.         "s": "ss"
224.     }
225.     # 解析参数
226.     expression = expression[2:-1]
227.     # 判断其开头，截取尾部
228.     min_unit = None
229.     for key, value in format_units.items():
230.         if key in expression:
231.             min_unit = key
232.             expression = expression.replace(key, value)
234.     # 替换完毕，确定是否有数字
235.     logging.info(f"转换为Python格式的表达式：expression: {expression}")
236.     # 定义正则表达式模式
237.     pattern = r'([^0-9]+)([-+]\d+(\*\d+)?)(?:_([YMdHms]))?'
238.     matches = re.match(pattern, expression)
239.     # 输出拆分结果
240.     if matches:
241.         date_part = matches.group(1)
242.         remainder = matches.group(2)
243.         unit = matches.group(4)
244.         if unit is not None and unit in unit_map.keys():
245.             min_unit = unit_map[unit]
246.         return calculate_expression(min_unit, date_part, remainder)
247.     else:
248.         return expression
251. def calculate_expression(min_unit: str, date_part: str, remainder: str) -> str:
252.     """
253.     计算表达式
254.     :param min_unit: 最小单位
255.     :param date_part: 日期表达式部分
256.     :param remainder: 偏移量部分
257.     :return: 计算之后的结果
258.     """
259.     logging.info(f"开始计算表达式：min_unit: {min_unit}, date_part: {date_part}, remainder:{remainder}")
260.     # 获取当前日期和时间
261.     now = datetime.now()
262.     # 计算时间的偏移量
263.     if remainder is None:
264.         # 格式化的日期
265.         formatted_datetime = now.strftime(date_part)
266.         logging.info(f"日期偏移量为空，返回值:{formatted_datetime}")
267.         return formatted_datetime
268.     else:
269.         # 计算偏移量
270.         plus_or_sub = remainder[0:1]
271.         offset = eval(remainder[1:])
272.         logging.info(f"计算偏移量，plus_or_sub：{plus_or_sub}， offset：{offset}")
273.         if min_unit == 'yyyy':
274.             if plus_or_sub == '-':
275.                 now = now - relativedelta(years=offset)
276.             else:
277.                 now = now + relativedelta(years=offset)
278.         elif min_unit == 'MM':
279.             if plus_or_sub == '-':
280.                 now = now - relativedelta(months=offset)
281.             else:
282.                 now = now + relativedelta(months=offset)
283.         elif min_unit == 'dd':
284.             if plus_or_sub == '-':
285.                 now = now - relativedelta(days=offset)
286.             else:
287.                 now = now + relativedelta(days=offset)
288.         elif min_unit == 'HH':
289.             if plus_or_sub == '-':
290.                 now = now - relativedelta(hours=offset)
291.             else:
292.                 now = now + relativedelta(hours=offset)
293.         elif min_unit == 'mm':
294.             if plus_or_sub == '-':
295.                 now = now - relativedelta(minutes=offset)
296.             else:
297.                 now = now + relativedelta(minutes=offset)
298.         elif min_unit == 'ss':
299.             if plus_or_sub == '-':
300.                 now = now - relativedelta(seconds=offset)
301.             else:
302.                 now = now + relativedelta(seconds=offset)
303.         formatted_datetime = now.strftime(date_part)
304.         logging.info(f"日期偏移量为空，返回值:{formatted_datetime}")
305.         return formatted_datetime
308. def general_reader(json_config: json) -> json:
309.     """
310.     生成配置的reader部分
311.     :param json_config: 配置
312.     :return: JSON结果
313.     """
314.     logging.info(f"开始生成DataX的配置JSON文件的reader内容")
315.     reader_json = json.loads("{}", encoding='utf-8')
316.     reader_json['name'] = "mysqlreader"
317.     reader_json['parameter'] = {}
318.     reader_json['parameter']['username'] = json_config['datasource']['username']
319.     reader_json['parameter']['password'] = json_config['datasource']['password']
320.     reader_json['parameter']['column'] = json_config['table']['column']
321.     reader_json['parameter']['connection'] = [{}]
322.     reader_json['parameter']['connection'][0]['table'] = json_config['table']['table']
323.     reader_json['parameter']['connection'][0]['jdbcUrl'] = [general_jdbc_url(json_config)]
324.     where_sql = json_config['table']['where']
325.     if where_sql is not None and where_sql != '':
326.         reader_json['parameter']['where'] = parse_where_sql(where_sql)
327.     return reader_json
330. def general_writer(json_config: json) -> json:
331.     """
332.     生成配置的Writer部分
333.     :param json_config: 配置
334.     :return: JSON结果
335.     """
336.     columns = json_config['table']['column']
337.     new_columns = []
338.     for column in columns:
339.         column = str(column).replace("`", "")
340.         if " AS " in str(column).upper():
341.             new_columns.append(str(column).split(" AS ")[1].strip())
342.         else:
343.             new_columns.append(str(column).strip())
344.     logging.info(f"开始生成DataX的配置JSON文件的Writer内容")
345.     writer_json = json.loads("{}", encoding='utf-8')
346.     writer_json['name'] = "kafkawriter"
347.     writer_json['parameter'] = {}
348.     writer_json['parameter']['bootstrapServers'] = "IP:19092,IP:19093,IP:19094"
349.     writer_json['parameter']['topic'] = json_config['kafka']['topic']
350.     writer_json['parameter']['ack'] = "all"
351.     writer_json['parameter']['batchSize'] = 1000
352.     writer_json['parameter']['retries'] = 3
353.     writer_json['parameter']['keySerializer'] = "org.apache.kafka.common.serialization.StringSerializer"
354.     writer_json['parameter']['valueSerializer'] = "org.apache.kafka.common.serialization.StringSerializer"
355.     writer_json['parameter']['fieldDelimiter'] = ","
356.     writer_json['parameter']['writeType'] = "json"
357.     writer_json['parameter']['topicNumPartition'] = 1
358.     writer_json['parameter']['topicReplicationFactor'] = 1
359.     writer_json['parameter']['encryptionKey'] = "5s8FGjerddfWkG/b64CGHHZYvQ=="
360.     writer_json['parameter']['column'] = new_columns
361.     return writer_json
364. def general_datax_job_config(datax_config: str):
365.     """
366.     生成job的配置内容
367.     :param datax_config: 配置
368.     :return: 完整的JSON内容
369.     """
370.     logging.info(f"开始生成DataX的配置JSON文件内容, {datax_config}")
371.     json_config = json.loads(datax_config, encoding='utf-8')
372.     # 判定是否需要查询表
373.     datasource = json_config['datasource']
374.     table = json_config['table']['table']
375.     search_table_sql_list = json_config['table']['searchTableSql']
376.     if search_table_sql_list is not None and len(search_table_sql_list) > 0:
377.         # 查询表列表，覆盖原来的配置信息
378.         table_list = search_table_list(datasource, search_table_sql_list)
379.     else:
380.         table_list = [table]
381.     json_config['table']['table'] = table_list
383.     # 开始生成配置文件
384.     job_json = general_default_job_config()
385.     job_json['job']['content'][0]['reader'] = general_reader(json_config)
386.     job_json['job']['content'][0]['writer'] = general_writer(json_config)
387.     return job_json
390. def write_job_file(base_path: str, job_config: json) -> str:
391.     """
392.     生成job的JSON配置文件
393.     :param base_path: 根路径
394.     :param job_config: 配置信息
395.     :return: 完整的JSON文件路径
396.     """
397.     # 生成一个脚本
398.     logging.info(f"开始创建DataX的配置JSON文件")
399.     date_day = datetime.now().strftime('%Y-%m-%d')
400.     timestamp_milliseconds = int(datetime.now().timestamp() * 1000)
401.     # 生成UUID
402.     file_name = str(uuid.uuid4()).replace("-", "") + "_" + str(timestamp_milliseconds) + ".json"
403.     # 完整文件路径
404.     # 创建文件夹
405.     mkdir_if_not_exist(base_path + "/task/datax/json/" + date_day)
406.     complex_file_path = base_path + "/task/datax/json/" + date_day + "/" + file_name
407.     logging.info(f"完整的DataX的配置JSON文件路径:{complex_file_path}")
408.     with open(complex_file_path, 'w+', encoding='utf-8') as f:
409.         f.write(json.dumps(job_config, ensure_ascii=False))
410.     return complex_file_path
413. def mkdir_if_not_exist(path):
414.     """
415.     创建目录
416.     :param path: 目录路径
417.     :return: None
418.     """
419.     os.makedirs(path, exist_ok=True)
422. def write_task_file(base_path: str, python_path: str, datax_path: str, job_file_path: str) -> str:
423.     """
424.     写shell脚本文件
425.     :param base_path: 跟路径
426.     :param python_path: python执行文件路径
427.     :param datax_path: datax执行文件路径
428.     :param job_file_path: JSON配置文件路径
429.     :return: shell脚本的完整路径
430.     """
431.     # 组合内容
432.     logging.info(f"开始创建Shell脚本文件")
433.     task_content = python_path + " " + datax_path + " " + job_file_path
434.     # 生成一个脚本
435.     date_day = datetime.now().strftime('%Y-%m-%d')
436.     timestamp_milliseconds = int(datetime.now().timestamp() * 1000)
437.     # 生成UUID
438.     task_file_name = str(uuid.uuid4()).replace("-", "") + "_" + str(timestamp_milliseconds) + ".sh"
439.     # 完整文件路径
440.     # 创建文件夹
441.     mkdir_if_not_exist(base_path + "/task/datax/shell/" + date_day)
442.     complex_file_path = base_path + "/task/datax/shell/" + date_day + "/" + task_file_name
443.     logging.info(f"完整的shell脚本路径: {complex_file_path}")
444.     with open(complex_file_path, 'w+', encoding='utf-8') as f:
445.         f.write(task_content)
446.     # 添加执行权限
447.     current_permissions = os.stat(complex_file_path).st_mode
448.     # 添加执行权限 (权限值 0o111 表示用户、组和其他人的执行权限)
449.     new_permissions = current_permissions | 0o111
450.     # 使用 os.chmod 设置新的权限
451.     os.chmod(complex_file_path, new_permissions)
452.     return complex_file_path
455. def signs(dd_secret: str, timestamp: str) -> str:
456.     """
457.     钉钉机器人签名
458.     :param dd_secret: 秘钥
459.     :param timestamp: 时间戳
460.     :return: 签名
461.     """
462.     secret_enc = dd_secret.encode('utf-8')
463.     string_to_sign = '{}\n{}'.format(timestamp, dd_secret)
464.     string_to_sign_enc = string_to_sign.encode('utf-8')
465.     hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
466.     sign = urllib.parse.quote(base64.b64encode(hmac_code))
467.     return sign
470. def real_send_msg(dd_secret: str, dd_access_token: str, text: json):
471.     """
472.     发送钉钉机器人消息
473.     :param dd_secret: 秘钥
474.     :param dd_access_token: token
475.     :param text: 内容
476.     :return: None
477.     """
478.     timestamp = str(round(time.time() * 1000))
479.     sign = signs(dd_secret, timestamp)
480.     headers = {'Content-Type': 'application/json'}
481.     web_hook = f'https://oapi.dingtalk.com/robot/send?access_token={dd_access_token}&timestamp={timestamp}&sign={sign}'
482.     # 定义要发送的数据
483.     requests.post(web_hook, data=json.dumps(text), headers=headers)
486. def send_msg(dd_secret: str, dd_access_token: str, job_start_time: str, total_count: int, success_count: int, fail_task_list: List[str]):
487.     """
488.     组合钉钉消息
489.     :param dd_secret: 秘钥
490.     :param dd_access_token: token
491.     :param job_start_time: 任务开始时间
492.     :param total_count: 总任务数
493.     :param success_count: 成功任务数
494.     :return: NONE
495.     """
496.     title = '### <font color=#CCCC00>数据同步结果</font>'
497.     if success_count == total_count:
498.         title = '### <font color=#00FF00>数据同步结果</font>'
499.     elif success_count == 0:
500.         title = '### <font color=#FF0000>数据同步结果</font>'
502.     end_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
503.     result = {
504.         "msgtype": "markdown",
505.         "markdown": {
506.             "title": "数据同步结果",
507.             "text": title + ' \n\n\n\n- '
508.                     + "总同步任务数:" + str(total_count) + "\n\n- "
509.                     + "成功任务数:" + str(success_count) + "\n\n- "
510.                     + "失败任务数" + str(total_count - success_count) + "\n\n- "
511.                     + "开始时间:" + str(job_start_time) + "\n\n- "
512.                     + "结束时间:" + str(end_time) + "\n\n- "
513.                     + "失败列表:" + str(fail_task_list) + "\n\n "
514.         }
515.     }
516.     if success_count < total_count:
517.         result['markdown']['at'] = json.loads("{"atMobiles": ["12345678997"]}")
518.     real_send_msg(dd_secret, dd_access_token, result)
521. def run_job(dd_secret, dd_access_token, job_start_time, base_path: str, python_script_path: str, datax_json_path: str):
522.     """
523.     运行任务
524.     :param dd_secret: 秘钥
525.     :param dd_access_token: token
526.     :param job_start_time: 任务开始时间
527.     :param base_path: 根路径
528.     :param python_script_path: Python执行路径
529.     :param datax_json_path: datax执行路径
530.     :return: NONE
531.     """
532.     task_content_list = read_all_files_in_directory(base_path + "/task/config/")
533.     success_count = 0
534.     total_count = len(task_content_list)
535.     fail_task_list = []
536.     for task_content in task_content_list:
537.         try:
538.             logging.info(f"开始生成，配置文件名称：{task_content}")
539.             job_config = general_datax_job_config(task_content_list[task_content])
540.             job_file_path = write_job_file(base_path, job_config)
541.             shell_path = write_task_file(base_path, python_script_path, datax_json_path, job_file_path)
542.             logging.info(f"shell脚本创建成功，路径为：{base_path}")
543.             # 调用脚本
544.             call_shell(shell_path)
545.             success_count += 1
546.         except Exception as e:
547.             fail_task_list.append(task_content)
548.             logging.error(f"配置文件：{task_content} 执行失败", e)
549.     # 发送消息
550.     send_msg(dd_secret, dd_access_token, job_start_time, total_count, success_count, fail_task_list)
553. def call_shell(shell_path: str):
554.     """
555.     执行shell脚本
556.     :param shell_path: shell脚本路径
557.     :return: NONE
558.     """
559.     logging.info(f"调用shell脚本，路径为：{shell_path}")
560.     result = subprocess.run(shell_path,
561.                             check=True,
562.                             shell=True,
563.                             universal_newlines=True,
564.                             stdout=subprocess.PIPE,
565.                             stderr=subprocess.PIPE)
567.     # 输出标准输出
568.     logging.info(f"shell脚本{shell_path}标准输出：%s", result.stdout)
569.     # # 输出标准错误输出
570.     logging.info(f"shell脚本{shell_path}标准错误输出：%s", result.stderr)
571.     # # 输出返回码
572.     logging.info(f"shell脚本{shell_path}的返回码：%s", result.returncode)
575. if __name__ == '__main__':
576.     """
577.     码中台数据同步任务脚本
578.     使用前请修改如下配置信息：
579.       - secret  钉钉机器人的秘钥
580.       - access_token  钉钉机器人的token
581.       - python_path   Python的安装路径
582.       - datax_path   datax的执行文件路径
583.     """
584.     # 钉钉配置
585.     start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
586.     secret = ''
587.     access_token = ''
588.     python_path = "/usr/bin/python3"
589.     datax_path = "/opt/datax-k/bin/datax.py"
590.     # 当前脚本文件的目录路径
591.     script_dir = '/opt/data-job'
592.     curr_date_day = datetime.now().strftime('%Y-%m-%d')
593.     # 创建文件夹
594.     mkdir_if_not_exist(script_dir + "/logs/" + curr_date_day)
595.     logging.basicConfig(level=logging.INFO,
596.                         format='%(asctime)s - %(levelname)s - %(lineno)d - %(message)s',
597.                         filename='logs/' + curr_date_day + '/app.log',
598.                         filemode='w')
599.     run_job(secret, access_token, start_time, script_dir, python_path, datax_path)
600.     logging.shutdown()

复制代码

• 同步日期的控制
  

我们在之前的任务同步中，遇到的问题便是日期的修改很贫苦，因此我们需要一个更加简朴的方式来举行日期的批量更新。在我们上面的调度脚本中，包含了对日期表达式的解析，我们自定义了一种时间的表达式$[yyyyMMddHHmmss+/-N_Y] 通过解析该表达式，我们可以生成需要的任意时间，该时间表达式的含义为：

• yyyy 表示年份
  
• MM 表示月份
  
• dd 表示日期
  
• HH  表示24进制小时
  
• mm  表示分钟
  
• ss  表示秒
  
• 
  
  
  
  • 表示当前时间加上N
    
  
  
• 
  
  
  
  • 表示当前时间减去N
    
  
  
• _Y  表示加减的单位，可以是YMdHms(年、月、日、时、分、秒)
  

通过对该表达式的解析，我们可以生成相对于当前之前或之后的任何格式的时间字符串，将其用于同步的where条件中，既可以完成针对时间的解析。

• 怎样更新日期
  

日期现在可以盘算，但是我们需要能够批量修改设置文件中的WHERE条件中的时间表达式，如我们想同步8天前的数据，我们就需要将脚本中的表达式修改为$[yyyyMMdd-8_d] ,即代表当前时间减去8天，这样我们就可以同步八天前那一天的数据，但是我们可能想同步从8天气到现在的所有数据，那么我们希望我们也能批量修改where表达式中的条件，如将=改为>=。
鉴于以上的需求，我们开辟了一个新的Python脚本，通过简朴的设置，即可一次修改所有脚本中的where条件中的表达式，这样，我们只需要执行两个脚本，就完成了统统，再也不需要依次修改执行10个工作流了。
[code]import jsonimport osimport loggingfrom typing import List, Mappingimport refrom datetime import datetime, datedef list_files_in_directory(directory_path: str) -> List[str]:    """    获取目录下的所有以.json结尾的文件    :param directory_path: 目录    :return: 文件列表    """    entries = os.listdir(directory_path)    # 过滤出所有文件    files = [entry for entry in entries if             os.path.isfile(os.path.join(directory_path, entry))              and entry.endswith(".json")]    logging.info(f"读取设置文件数量：{len(files)}")    return filesdef read_file_content(file_path: str) -> str:    """    读取文件内容    :param file_path: 文件路径    :return: 文件内容    """    with open(file_path, 'r', encoding='utf-8') as file:        content = file.read()    return contentdef read_all_files_in_directory(directory_path: str) -> Mapping[str, str]:    """    读取文件夹下面的所有文件的内容    :param directory_path: 文件夹路径    :return: 内容map    """    logging.info(f"开始读取所有的设置文件信息")    files = list_files_in_directory(directory_path)    file_contents = {}    for file in files:        file_path = os.path.join(directory_path, file)        content = read_file_content(file_path)        file_contents[file] = content    sorted_items = sorted(file_contents.items())    sorted_dict = dict(sorted_items)    return file_contentsdef parse_where_sql(where_sql: str, sub_day: int, comparator: str = None) -> str:    """    解析where语句    :param where_sql: 原始where语句    :param sub_day: 天数    :param comparator: 比较符  包罗 = != > < >=