FlinkX的安装与简单使用

Posted lmandcc

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了FlinkX的安装与简单使用相关的知识,希望对你有一定的参考价值。

FlinkX的安装与简单使用

FlinkX的安装

安装unzip:yum install unzip

1、上传并解压

unzip flinkx-1.10.zip -d /usr/local/soft/

2、配置环境变量

3、给bin/flinkx这个文件加上执行权限

chmod a+x flinkx

4、修改配置文件,设置运行端口

vim flinkconf/flink-conf.yaml
## web服务端口,不指定的话会随机生成一个
rest.bind-port: 8888

FlinkX的简单使用

mysqlToHDFS

  • 配置文件
{
    "job": {
        "content": [
            {
                "reader": {
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://master:3306/student?characterEncoding=utf8"
                                ],
                                "table": [
                                    "student"
                                ]
                            }
                        ],
                        "column": [
                            "*"
                        ],
                        "customSql": "",
                        "where": "clazz = \'理科二班\'",
                        "splitPk": "",
                        "queryTimeOut": 1000,
                        "requestAccumulatorInterval": 2
                    },
                    "name": "mysqlreader"
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "path": "hdfs://master:9000/data/flinkx/student",
                        "defaultFS": "hdfs://master:9000",
                        "column": [
                            {
                                "name": "col1",
                                "index": 0,
                                "type": "string"
                            },
                            {
                                "name": "col2",
                                "index": 1,
                                "type": "string"
                            },
                            {
                                "name": "col3",
                                "index": 2,
                                "type": "string"
                            },
                            {
                                "name": "col4",
                                "index": 3,
                                "type": "string"
                            },
                            {
                                "name": "col5",
                                "index": 4,
                                "type": "string"
                            },
                            {
                                "name": "col6",
                                "index": 5,
                                "type": "string"
                            }
                        ],
                        "fieldDelimiter": ",",
                        "fileType": "text",
                        "writeMode": "overwrite"
                    }
                }
            }
        ],
        "setting": {
            "restore": {
                "isRestore": false,
                "isStream": false
            },
            "errorLimit": {},
            "speed": {
                "channel": 1
            }
        }
    }
}
  • 启动任务
flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHDFS.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/
  • 监听日志

flinkx 任务启动后,会在执行命令的目录下生成一个nohup.out文件

tail -f nohup.out
  • 通过web界面查看任务运行情况
http://master:8888

MySQLToHive

  • 配置文件
{
    "job": {
        "content": [
            {
                "reader": {
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://master:3306/student?characterEncoding=utf8"
                                ],
                                "table": [
                                    "student"
                                ]
                            }
                        ],
                        "column": [
                            "*"
                        ],
                        "customSql": "",
                        "where": "clazz = \'文科二班\'",
                        "splitPk": "id",
                        "queryTimeOut": 1000,
                        "requestAccumulatorInterval": 2
                    },
                    "name": "mysqlreader"
                },
                "writer": {
                    "name": "hivewriter",
                    "parameter": {
                        "jdbcUrl": "jdbc:hive2://master:10000/testflinkx",
                        "username": "",
                        "password": "",
                        "fileType": "text",
                        "fieldDelimiter": ",",
                        "writeMode": "overwrite",
                        "compress": "",
                        "charsetName": "UTF-8",
                        "maxFileSize": 1073741824,
                        "tablesColumn": "{\\"student\\":[{\\"key\\":\\"id\\",\\"type\\":\\"string\\"},{\\"key\\":\\"name\\",\\"type\\":\\"string\\"},{\\"key\\":\\"age\\",\\"type\\":\\"string\\"}]}",
                        "defaultFS": "hdfs://master:9000"
                    }
                }
            }
        ],
        "setting": {
            "restore": {
                "isRestore": false,
                "isStream": false
            },
            "errorLimit": {},
            "speed": {
                "channel": 3
            }
        }
    }
}
  • 在hive中创建testflinkx数据库,并创建student分区表
create database testflinkx;
use testflinkx;
CREATE TABLE `student`(
  `id` string, 
  `name` string, 
  `age` string)
PARTITIONED BY ( 
  `pt` string)
ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY \',\'
  • 启动hiveserver2
# 第一种方式:
hiveserver2
# 第二种方式:
hive --service hiveserver2
  • 启动任务
flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHive.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/
  • 查看日志及运行情况同上

MySQLToHBase

  • 配置文件
{
    "job": {
        "content": [
            {
                "reader": {
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://master:3306/student?characterEncoding=utf8"
                                ],
                                "table": [
                                    "score"
                                ]
                            }
                        ],
                        "column": [
                            "*"
                        ],
                        "customSql": "",
                        "splitPk": "student_id",
                        "queryTimeOut": 1000,
                        "requestAccumulatorInterval": 2
                    },
                    "name": "mysqlreader"
                },
                "writer": {
                    "name": "hbasewriter",
                    "parameter": {
                        "hbaseConfig": {
                            "hbase.zookeeper.property.clientPort": "2181",
                            "hbase.rootdir": "hdfs://master:9000/hbase",
                            "hbase.cluster.distributed": "true",
                            "hbase.zookeeper.quorum": "master,node1,node2",
                            "zookeeper.znode.parent": "/hbase"
                        },
                        "table": "testFlinkx",
                        "rowkeyColumn": "$(cf1:student_id)_$(cf1:course_id)",
                        "column": [
                            {
                                "name": "cf1:student_id",
                                "type": "string"
                            },
                            {
                                "name": "cf1:course_id",
                                "type": "string"
                            },
                            {
                                "name": "cf1:score",
                                "type": "string"
                            }
                        ]
                    }
                }
            }
        ],
        "setting": {
            "restore": {
                "isRestore": false,
                "isStream": false
            },
            "errorLimit": {},
            "speed": {
                "channel": 3
            }
        }
    }
}
  • 启动hbase 并创建testflinkx表
create \'testFlinkx\',\'cf1\'
  • 启动任务
flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHBase.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/
  • 查看日志及运行情况同上

MySQLToMySQL

  • 配置文件
{
    "job": {
      "content": [
        {
          "reader": {
            "name": "mysqlreader",
            "parameter": {
              "column": [
                {
                  "name": "id",
                  "type": "int"
                },
                {
                  "name": "name",
                  "type": "string"
                },
                {
                  "name": "age",
                  "type": "int"
                },
                {
                  "name": "gender",
                  "type": "string"
                },
                {
                  "name": "clazz",
                  "type": "string"
                }
              ],
              "username": "root",
              "password": "123456",
              "connection": [
                {
                  "jdbcUrl": [
                    "jdbc:mysql://master:3306/student?useSSL=false"
                  ],
                  "table": [
                    "student"
                  ]
                }
              ]
            }
          },
          "writer": {
            "name": "mysqlwriter",
            "parameter": {
              "username": "root",
              "password": "123456",
              "connection": [
                {
                  "jdbcUrl": "jdbc:mysql://master:3306/student?useSSL=false",
                  "table": [
                    "student2"
                  ]
                }
              ],
              "writeMode": "insert",
              "column": [
                {
                    "name": "id",
                    "type": "int"
                  },
                  {
                    "name": "name",
                    "type": "string"
                  },
                  {
                    "name": "age",
                    "type": "int"
                  },
                  {
                    "name": "gender",
                    "type": "string"
                  },
                  {
                    "name": "clazz",
                    "type": "string"
                  }
              ]
            }
          }
        }
      ],
      "setting": {
        "speed": {
          "channel": 1,
          "bytes": 0
        }
      }
    }
  }

Flinkx-web安装

1.Flinkx的安装

  • 1.下载源码包

Flinkx-web安装_bc

Flinkx下载地址

2.解压和安装

2.1 用解压软件进行解压即可

2.2 window环境下双击解压后的flinkx-1.8_release\\bin\\install_jars.bat

​ linux/mac环境下执行 sh install_jars.sh

本地安装db2jcc,ojdbc8,gbase,dm7的jdbc驱动的JAR包

3.编译插件

使用cmd命令进入到 flinkx-1.8_release目录,使用maven命令进行编译

mvn clean package -Dmaven.test.skip=true

Flinkx-web安装_jar_02

4.配置文件

4.1 flinkx-1.8_release\\flinkconf目录下flink-conf.yaml里添加如下:

rest.bind-port: 8888

4.2 flinkx-1.8_release\\下面创建job目录

4.3 在 flinkx-1.8_release\\job目录下面添加stream.json


"job" :
"content" : [
"reader":
"name": "streamreader",
"parameter":
"column": [

"name": "id",
"type": "int"
,

"name": "name",
"type": "string"

]

,
"writer" :
"parameter" :
"print": false
,
"name" : "streamwriter"

],
"setting" :
"restore" :
"isRestore" : false,
"isStream" : false
,
"errorLimit" :
,
"speed" :
"bytes" : 0,
"channel" : 1,
"rebalance" : true



4.4将图中所选择的文件上传到Linux服务器

Flinkx-web安装_bc_03

4.5编写启动脚本start.sh文件

sh ./bin/flinkx-job ./job/stream.json -flinkconf ./flinkconf/ -pluginRoot ./plugins/

5.执行任务:

[root@mini4 hadoop]# tar -zxvf flinkx-1.8_release.tar.gz

[root@mini4 flinkx-1.8_release]# sh start.sh

演示结果如下

Flinkx-web安装_flink_04


以上是关于FlinkX的安装与简单使用的主要内容,如果未能解决你的问题,请参考以下文章

带你详解数栈FlinkX实时采集原理与使用

FlinkX快速开始(一)

数据集成工具的使用---FlinkX 从理论学习到熟练使用

好消息!数栈FlinkX技术团队将FlinkX开源项目同步推送到Gitee啦!

在Tomcat的安装目录下conf目录下的server.xml文件中增加一个xml代码片段,该代码片段中每个属性的含义与用途

GitLab安装与汉化-实战