登录 注册
开源 企业版 高校版 私有云 Gitee AI NEW
开源项目 > 服务器应用 > 大数据 &&
扫描微信二维码支付
取消
支付完成
Watch
不关注 关注所有动态 仅关注版本发行动态 关注但不提醒动态
48 Star 499 Fork 229

GVP袋鼠云 / chunjun

代码 Issues 13 Pull Requests 1 Wiki 统计 流水线
服务
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目，私有仓库也完全免费 ：）
免费加入
克隆/下载
phoenixreader.md 14.38 KB
一键复制 编辑 原始数据 按行查看 历史
tudou 提交于 2021-02-05 18:05 . 1、新增Oracle LogMiner实时采集插件

Phoenix Reader

一、插件名称

名称：phoenixreader

二、支持的数据源版本

phoenix4.12.0-HBase-1.3及以上

三、参数说明

  • connection
    • 描述：数据库连接参数，包含jdbcUrl、schema、table等参数
    • 必选：是
    • 字段类型：List
      • 示例：指定jdbcUrl、schema、table
    "connection": [{
     "jdbcUrl": ["jdbc:phoenix:node01,node02,node03:2181"],
     "table": ["table"],
     "schema":"public"
    }]
    • 默认值：无

  • schema
    • 描述：数据库schema名
    • 必选：否
    • 字段类型：String
    • 默认值：无

  • table

    • 描述：目的表的表名称。目前只支持配置单个表，后续会支持多表
    • 必选：是
    • 字段类型：List
    • 默认值：无

  • jdbcUrl

    • 描述：针对关系型数据库的jdbc连接字符串
      jdbcUrl参考文档：phoniex官方文档
    • 必选：是
    • 字段类型：List
    • 默认值：无

  • username
    • 描述：数据源的用户名
    • 必选：是
    • 字段类型：String
    • 默认值：无

  • password
    • 描述：数据源指定用户名的密码
    • 必选：是
    • 字段类型：String
    • 默认值：无

  • where
    • 描述：筛选条件，reader插件根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > time。
    • 注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。
    • 必选：否
    • 字段类型：String
    • 默认值：无

  • splitPk
    • 描述：当speed配置中的channel大于1时指定此参数，Reader插件根据并发数和此参数指定的字段拼接sql，使每个并发读取不同的数据，提升读取速率。
    • 注意:
      • 推荐splitPk使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。
      • 目前splitPk仅支持整形数据切分，不支持浮点、字符串、日期等其他类型。如果用户指定其他非支持类型，FlinkX将报错！
      • 如果channel大于1但是没有配置此参数，任务将置为失败。
    • 必选：否
    • 字段类型：String
    • 默认值：无

  • fetchSize
    • 描述：一次性从数据库中读取多少条数据，jdbc默认一次将所有结果都读取到内存中，在数据量很大时可能会造成OOM，设置这个参数可以控制每次读取fetchSize条数据。
    • 注意：此参数的值不可设置过大，否则会读取超时，导致任务失败。
    • 必选：否
    • 字段类型：int
    • 默认值：1000

  • queryTimeOut
    • 描述：查询超时时间，单位秒。
    • 注意：当数据量很大，或者从视图查询，或者自定义sql查询时，可通过此参数指定超时时间。
    • 必选：否
    • 字段类型：int
    • 默认值：1000

  • customSql
    • 描述：自定义的查询语句，如果只指定字段不能满足需求时，可通过此参数指定查询的sql，可以是任意复杂的查询语句。
    • 注意：
      • 只能是查询语句，否则会导致任务失败；
      • 查询语句返回的字段需要和column列表里的字段对应；
      • 当指定了此参数时，connection里指定的table无效；
      • 当指定此参数时，column必须指定具体字段信息，不能以*号代替；
    • 必选：否
    • 字段类型：String
    • 默认值：无

  • column
    • 描述：需要读取的字段。
    • 格式：支持3种格式
      1.读取全部字段，如果字段数量很多，可以使用下面的写法：
"column":["*"]

2.只指定字段名称：

"column":["id","name"]

3.指定具体信息：

"column": [{
    "name": "col",
    "type": "datetime",
    "format": "yyyy-MM-dd hh:mm:ss",
    "value": "value"
}]
  • 属性说明:
    • name：字段名称
    • type：字段类型，可以和数据库里的字段类型不一样，程序会做一次类型转换
    • format：如果字段是时间字符串，可以指定时间的格式，将字段类型转为日期格式返回
    • value：如果数据库里不存在指定的字段，则会把value的值作为常量列返回，如果指定的字段存在，当指定字段的值为null时，会以此value值作为默认值返回
  • 必选：是
  • 字段类型：List
  • 默认值：无

  • polling
    • 描述：是否开启间隔轮询，开启后会根据pollingInterval轮询间隔时间周期性的从数据库拉取数据。开启间隔轮询还需配置参数pollingInterval，increColumn，可以选择配置参数startLocation。若不配置参数startLocation，任务启动时将会从数据库中查询增量字段最大值作为轮询的开始位置。
    • 必选：否
    • 字段类型：Boolean
    • 默认值：false

  • pollingInterval
    • 描述：轮询间隔时间，从数据库中拉取数据的间隔时间，默认为5000毫秒。
    • 必选：否
    • 字段类型：long
    • 默认值：5000

  • increColumn

    • 增量字段，可以是对应的增量字段名，也可以是纯数字，表示增量字段在column中的顺序位置（从0开始）
    • 必选：否
    • 字段类型：String或int
    • 默认值：无

  • startLocation

    • 描述：增量查询起始位置
    • 必选：否
    • 字段类型：String
    • 默认值：无

  • useMaxFunc

    • 描述：用于标记是否保存endLocation位置的一条或多条数据，true：不保存，false(默认)：保存， 某些情况下可能出现最后几条数据被重复记录的情况，可以将此参数配置为true
    • 必选：否
    • 字段类型：Boolean
    • 默认值：false

  • requestAccumulatorInterval

    • 描述：发送查询累加器请求的间隔时间。
    • 必选：否
    • 字段类型：int
    • 默认值：2

四、配置示例

1、基础配置

{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "",
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}

2、多通道

{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "",
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 3,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}

3、指定customSql

{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "select * from tableTest",
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}
{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "",
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}

4、增量同步指定customSql

{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "increColumn": "id",
          "startLocation": "20",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "",
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}

5、增量轮询

{
  "job": {
    "content": [{
      "reader": {
        "parameter" : {
          "column" : [ {
            "name" : "id",
            "type" : "bigint"
          }, {
            "name" : "user_id",
            "type" : "bigint"
          }, {
            "name" : "name",
            "type" : "varchar"
          } ],
          "username" : "",
          "password" : "",
          "connection" : [ {
            "jdbcUrl" : [ "jdbc:phoenix:node01,node02,node03:2181" ],
            "table" : [ "tableTest" ]
          } ],
          "where": "id > 1",
          "splitPk": "id",
          "fetchSize": 1000,
          "queryTimeOut": 1000,
          "customSql": "",
          "polling": true,
          "pollingInterval": 3000,
          "requestAccumulatorInterval": 2
        },
        "name" : "phoenixreader"
      },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": 0
      },
      "errorLimit": {
        "record": 100
      },
      "restore": {
        "maxRowNumForCheckpoint": 0,
        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      },
      "log" : {
        "isLogger": false,
        "level" : "debug",
        "path" : "",
        "pattern":""
      }
    }
  }
}
Java
1
https://gitee.com/dtstack_dev_0/chunjun.git
git@gitee.com:dtstack_dev_0/chunjun.git
dtstack_dev_0
chunjun
chunjun
1.10_release
点此查找更多帮助

搜索帮助

Git 命令在线学习 如何在 Gitee 导入 GitHub 仓库
Git 仓库基础操作
企业版和社区版功能对比
SSH 公钥设置
如何处理代码冲突
仓库体积过大，如何减小？
如何找回被删除的仓库数据
Gitee 产品配额说明
GitHub仓库快速导入Gitee及同步更新
什么是 Release（发行版）
将 PHP 项目自动发布到 packagist.org
评论
仓库举报
回到顶部