FlinkX配置文件解析

我不是女神ヾ 2023-02-12 05:44 9阅读 0赞

# FlinkX配置文件解析 #

## 配置文件 （本章节引自官网） ##

一个完整的Flinkx任务脚本配置包含 content， setting两个部分。content用于配置任务的输入源与输出源，其中包含reader，writer。而setting则配置任务整体的环境设定，其中包含restore，speed，errorLimit，dirty，log。具体如下所示：

{ 
    	"job" : { 
    		"content" :[{ 
        	"reader" : { 
    				......
          },
        	"writer" : { 
    				......
    			}
        }],
       "setting" : { 
          "restore" : { 
            ......
          },
          "speed" : { 
          	......
          },
          "errorLimit" : { 
          	......
          },
          "dirty" : { 
          	......
          },
          "log" : { 
          	......
          }
        }
    	}
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th></th> 
   <th>说明</th> 
   <th>是否必填</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>content</td> 
   <td>reader</td> 
   <td>reader插件详细配置</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>writer</td> 
   <td>writer插件详细配置</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>setting</td> 
   <td>restore</td> 
   <td>任务类型及断点续传配置</td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>speed</td> 
   <td>速率限制</td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>errorLimit</td> 
   <td>出错控制</td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>dirty</td> 
   <td>脏数据保存</td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>log</td> 
   <td>日志记录配置</td> 
   <td>否</td> 
  </tr> 
 </tbody> 
</table>

## content配置 ##

### reader ###

reader用于配置数据的输入源，即数据从何而来。具体配置如下所示：

"reader" : { 
      "name" : "xxreader",
      "parameter" : { 
    		......
      }
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>name</td> 
   <td>reader插件名称，具体名称参考各数据源配置文档</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>parameter</td> 
   <td>数据源配置参数，具体配置参考各数据源配置文档</td> 
   <td>是</td> 
  </tr> 
 </tbody> 
</table>

### writer ###

writer用于配置数据的输出源，即数据写往何处。具体配置如下所示：

"writer" : { 
      "name" : "xxwriter",
      "parameter" : { 
    		......
      }
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>name</td> 
   <td>writer插件名称，具体名称参考各数据源配置文档</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>parameter</td> 
   <td>数据源配置参数，具体配置参考各数据源配置文档</td> 
   <td>是</td> 
  </tr> 
 </tbody> 
</table>

## setting配置 ##

### restore ###

restore用于配置同步任务类型（离线同步、实时采集）和断点续传功能。具体配置如下所示：

"restore" : { 
      "isStream" : false,
      "isRestore" : false,
      "restoreColumnName" : "",
      "restoreColumnIndex" : 0,
      "maxRowNumForCheckpoint" : 10000
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
   <th>默认值</th> 
   <th>参数类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>isStream</td> 
   <td>是否为实时采集任务</td> 
   <td>否</td> 
   <td>false</td> 
   <td>Boolean</td> 
  </tr> 
  <tr> 
   <td>isRestore</td> 
   <td>是否开启断点续传</td> 
   <td>否</td> 
   <td>false</td> 
   <td>Boolean</td> 
  </tr> 
  <tr> 
   <td>restoreColumnName</td> 
   <td>断点续传字段名称</td> 
   <td>开启断点续传后必填</td> 
   <td>无</td> 
   <td>String</td> 
  </tr> 
  <tr> 
   <td>restoreColumnIndex</td> 
   <td>断点续传字段索引ID</td> 
   <td>开启断点续传后必填</td> 
   <td>-1</td> 
   <td>int</td> 
  </tr> 
  <tr> 
   <td>maxRowNumForCheckpoint</td> 
   <td>触发checkpoint数据条数</td> 
   <td>否</td> 
   <td>10000</td> 
   <td>int</td> 
  </tr> 
 </tbody> 
</table>

### speed ###

speed用于配置任务并发数及速率限制。具体配置如下所示：

"speed" : { 
      "channel": 1,
      "bytes": 0
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
   <th>默认值</th> 
   <th>参数类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>channel</td> 
   <td>任务并发数</td> 
   <td>否</td> 
   <td>1</td> 
   <td>int</td> 
  </tr> 
  <tr> 
   <td>bytes</td> 
   <td>bytes &gt;0则表示开启任务限速</td> 
   <td>否</td> 
   <td>Long.MAX_VALUE</td> 
   <td>Long</td> 
  </tr> 
 </tbody> 
</table>

### errorLimit ###

errorLimit用于配置任务运行时数据读取写入的出错控制。具体配置如下所示：

"errorLimit" : { 
      "record": 100,
      "percentage": 10.0
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
   <th>默认值</th> 
   <th>参数类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>record</td> 
   <td>错误阈值，当错误记录数超过此阈值时任务失败</td> 
   <td>否</td> 
   <td>0</td> 
   <td>int</td> 
  </tr> 
  <tr> 
   <td>percentage</td> 
   <td>错误比例阈值，当错误记录比例超过此阈值时任务失败</td> 
   <td>否</td> 
   <td>0.0</td> 
   <td>Double</td> 
  </tr> 
 </tbody> 
</table>

### dirty ###

dirty用于配置脏数据的保存，通常与上文出错控制联合使用。具体配置如下所示：

"dirty" : { 
      "path" : "xxx",
      "hadoopConfig" : { 
        ......
      }
     }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
   <th>默认值</th> 
   <th>参数类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>path</td> 
   <td>脏数据保存路径</td> 
   <td>是</td> 
   <td>无</td> 
   <td>Sring</td> 
  </tr> 
  <tr> 
   <td>hadoopConfig</td> 
   <td>Hadoop相关配置</td> 
   <td>是</td> 
   <td>无</td> 
   <td>K-V键值对</td> 
  </tr> 
 </tbody> 
</table>

参考模板如下：

"dirty" : { 
            "path" : "/user/hive/warehouse/xx.db/xx",
            "hadoopConfig" : { 
              "fs.default.name": "hdfs://0.0.0.0:9000",
              "dfs.ha.namenodes.ns1" : "nn1,nn2",
              "dfs.namenode.rpc-address.ns1.nn1" : "0.0.0.0:9000",
              "dfs.namenode.rpc-address.ns1.nn2" : "0.0.0.1:9000",
              "dfs.client.failover.proxy.provider.ns1" : "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
              "dfs.nameservices" : "ns1"
            }
          }

### log ###

log用于配置Flinkx中定义的插件日志的保存与记录。具体配置如下所示：

"log" : { 
      "isLogger": false,
      "level" : "info",
      "path" : "/tmp/dtstack/flinkx/",
      "pattern":""
    }

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>是否必填</th> 
   <th>默认值</th> 
   <th>参数类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>isLogger</td> 
   <td>是否保存日志记录</td> 
   <td>否</td> 
   <td>false</td> 
   <td>Boolean</td> 
  </tr> 
  <tr> 
   <td>level</td> 
   <td>日志级别</td> 
   <td>否</td> 
   <td>info</td> 
   <td>String</td> 
  </tr> 
  <tr> 
   <td>path</td> 
   <td>服务器上日志保存路径</td> 
   <td>否</td> 
   <td>/tmp/dtstack/flinkx/</td> 
   <td>String</td> 
  </tr> 
  <tr> 
   <td>pattern</td> 
   <td>日志输出格式</td> 
   <td>否</td> 
   <td>log4j：%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</td> 
   <td>String</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
   <td></td> 
   <td>logback : %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

# 示例 #

这是一个mongo => mysql 的实例

*  有四个分片，并且根据
 *  错误为无限制
 *  脏值保存在hadoop的上
 *  基本的reader 和 writer 配置

{ 
        "job": { 
            "setting": { 
                "speed": { 
                     "channel": 4,
                     "bytes": 1024
                },
                "errorLimit": { 
                    "record": 10000,
                    "percentage": 100
                },
                "dirty": { 
                    "path": "/tmp/flinkx/bond_info_mongodb_to_mysql/1",
                    "hadoopConfig": { 
                        "fs.default.name": "hdfs://flinkhadoop:8020",
                        "dfs.nameservices": "ns1",
                        "dfs.ha.namenodes.ns1": "flinkhadoop",
                        "dfs.namenode.rpc-address.ns1.nn1": "hdfs://flinkhadoop:8020",
                        "dfs.ha.automatic-failover.enabled": "true",
                        "dfs.client.failover.proxy.provider.ns1": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
                        "fs.hdfs.impl.disable.cache": "true"
                    }
                }
            },
            "content": [
                { 
                    "reader": { 
                        "parameter":{ 
                            "hostPorts":"xxx",
                            "username": "",
                            "password": "",
                            "database":"data",
                            "collectionName": "bond_info",
                            "fetchSize":100,
                            "column": [
                                { 
                                    "name":"bond_name",
                                    "type":"string"
                                },{ 
                                    "name":"bond_stop_time",
                                    "type":"string"
                                },{ 
                                    "name":"bond_time_limit",
                                    "type":"string"
                                },{ 
                                    "name":"bond_type",
                                    "type":"string"
                                }
                            ],
                            "filter": ""
                        },
                        "name":"mongodbreader"
                    },
                   "writer": { 
                        "name": "mysqlwriter",
                        "parameter": { 
                            "writeMode": "insert",
                            "username": "root",
                            "password": "xxxx",
                            "column": [
                                "bond_name",
                                "bond_stop_time",
                                "bond_time_limit",
                                "bond_type"
                            
                            ],
                            "batchSize": 1,
                            "session": [
                            ],
                            "connection": [
                                { 
                                    "jdbcUrl": "jdbc:mysql://xxxxx:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false",
                                    "table": [
                                        "bond_info"
                                    ]
                                }
                            ]
                        }
                    }
                }
            ]
        }
    }