{"id":148048,"date":"2020-08-30T18:02:10","date_gmt":"2020-08-30T10:02:10","guid":{"rendered":"http:\/\/4563.org\/?p=148048"},"modified":"2020-08-30T18:02:10","modified_gmt":"2020-08-30T10:02:10","slug":"flink-%e5%a6%82%e4%bd%95%e8%87%aa%e5%ae%9a%e4%b9%89mysql-source%e4%b8%8esink%ef%bc%8c%e5%ae%9e%e7%8e%b0mysql%e7%9a%84%e8%af%bb%e5%8f%96%e5%92%8c%e5%86%99%e5%85%a5","status":"publish","type":"post","link":"http:\/\/4563.org\/?p=148048","title":{"rendered":"Flink \u5982\u4f55\u81ea\u5b9a\u4e49mysql source\u4e0esink\uff0c\u5b9e\u73b0mysql\u7684\u8bfb\u53d6\u548c\u5199\u5165"},"content":{"rendered":"

<\/a>\u6570\u636e\u5e93\u8868\u51c6\u5907<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
<\/pre>\n<\/td>\n
\n
show<\/span> databases<\/span>;<\/span>
<\/span>
create<\/span> database<\/span> flink;<\/span>
<\/span>
use<\/span> flink<\/span>
<\/span>
CREATE<\/span> TABLE<\/span> `user`<\/span> (<\/span>
  `id`<\/span> int<\/span>(10<\/span>) unsigned<\/span> NOT<\/span> NULL<\/span> AUTO_INCREMENT COMMENT<\/span> '\u4e3b\u952eid'<\/span>,<\/span>
  `name`<\/span> varchar<\/span>(255<\/span>) NOT<\/span> NULL<\/span> DEFAULT<\/span> ''<\/span> COMMENT<\/span> '\u7528\u6237\u540d\u79f0'<\/span>,<\/span>
  `address`<\/span> varchar<\/span>(200<\/span>) NOT<\/span> NULL<\/span> DEFAULT<\/span> ''<\/span> COMMENT<\/span> '\u4f4f\u5740'<\/span>,<\/span>
  `sex`<\/span> tinyint<\/span>(1<\/span>) NOT<\/span> NULL<\/span> DEFAULT<\/span> '0'<\/span> COMMENT<\/span> '\u6027\u522b[0:\u672a\u77e5,1:\u7537,2:\u5973]'<\/span>,<\/span>
  `create_time`<\/span> timestamp<\/span> NOT<\/span> NULL<\/span> DEFAULT<\/span> CURRENT_TIMESTAMP<\/span> COMMENT<\/span> '\u521b\u5efa\u65f6\u95f4'<\/span>,<\/span>
  `update_time`<\/span> timestamp<\/span> NOT<\/span> NULL<\/span> DEFAULT<\/span> CURRENT_TIMESTAMP<\/span> ON<\/span> UPDATE<\/span> CURRENT_TIMESTAMP<\/span> COMMENT<\/span> '\u66f4\u65b0\u65f6\u95f4'<\/span>,<\/span>
  PRIMARY KEY<\/span> (`id`<\/span>) USING<\/span> BTREE<\/span>
) ENGINE<\/span>=InnoDB<\/span> AUTO_INCREMENT=1<\/span> DEFAULT<\/span> CHARSET<\/span>=utf8 COMMENT<\/span>='\u7528\u6237\u57fa\u7840\u4fe1\u606f\u8868'<\/span>;<\/span>
<\/span>
<\/span>
INSERT<\/span> INTO<\/span> user<\/span>  (name<\/span>, address,sex)  VALUES<\/span>  ('xiaoming'<\/span>,'Beijing'<\/span>,1<\/span>),('xiaoqiang'<\/span>,'GuangZhou'<\/span>,2<\/span>),('xiaohua'<\/span>,'Hubei'<\/span>,0<\/span>),('xiaoli'<\/span>,'GuangZhou'<\/span>,1<\/span>)<\/span>
<\/span>
<\/span>
select<\/span> * from<\/span> user<\/span>;<\/span>
+----+-----------+-----------+-----+---------------------+---------------------+<\/span><\/span>
| id | name      | address   | sex | create_time         | update_time         |<\/span>
+----+-----------+-----------+-----+---------------------+---------------------+<\/span><\/span>
| 1  | xiaoming  | Beijing   | 1   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 2  | xiaoqiang | GuangZhou | 2   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 3  | xiaohua   | Hubei     | 0   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 4  | xiaoli    | GuangZhou | 1   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
+----+-----------+-----------+-----+---------------------+---------------------+<\/span><\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>\u901a\u8fc7maven\u5728pom.xml\u4e2d\u6dfb\u52a0\u9a71\u52a8\u4f9d\u8d56<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
<\/pre>\n<\/td>\n
\n
<dependency><\/span>
    <groupId>mysql<\/groupId><\/span>
    <artifactId>mysql-connector-java<\/artifactId><\/span>
    <version>5.1.35<\/version><\/span>
<\/dependency><\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>\u5b9a\u4e49User\u5b9e\u4f53\u7c7b\uff0c\u7528\u4e8e\u5c01\u88c5\u6570\u636e<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
51<\/span>
52<\/span>
53<\/span>
54<\/span>
55<\/span>
56<\/span>
57<\/span>
58<\/span>
59<\/span>
60<\/span>
61<\/span>
62<\/span>
63<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.java.model;<\/span>
<\/span>
public<\/span> class<\/span> User<\/span> <\/span>{<\/span>
    public<\/span> int<\/span> id;<\/span>
    public<\/span> String name;<\/span>
    public<\/span> String address;<\/span>
    public<\/span> int<\/span> sex;<\/span>
<\/span>
    public<\/span> User<\/span>(int<\/span> id, String name, String address, int<\/span> sex)<\/span> <\/span>{<\/span>
        this<\/span>.id = id;<\/span>
        this<\/span>.name = name;<\/span>
        this<\/span>.address = address;<\/span>
        this<\/span>.sex = sex;<\/span>
    }<\/span>
<\/span>
    public<\/span> User<\/span>(String name, String address, int<\/span> sex)<\/span> <\/span>{<\/span>
        this<\/span>.name = name;<\/span>
        this<\/span>.address = address;<\/span>
        this<\/span>.sex = sex;<\/span>
    }<\/span>
<\/span>
    public<\/span> int<\/span> getId<\/span>()<\/span> <\/span>{<\/span>
        return<\/span> id;<\/span>
    }<\/span>
<\/span>
    public<\/span> void<\/span> setId<\/span>(int<\/span> id)<\/span> <\/span>{<\/span>
        this<\/span>.id = id;<\/span>
    }<\/span>
<\/span>
    public<\/span> String getName<\/span>()<\/span> <\/span>{<\/span>
        return<\/span> name;<\/span>
    }<\/span>
<\/span>
    public<\/span> void<\/span> setName<\/span>(String name)<\/span> <\/span>{<\/span>
        this<\/span>.name = name;<\/span>
    }<\/span>
<\/span>
    public<\/span> String getAddress<\/span>()<\/span> <\/span>{<\/span>
        return<\/span> address;<\/span>
    }<\/span>
<\/span>
    public<\/span> void<\/span> setAddress<\/span>(String address)<\/span> <\/span>{<\/span>
        this<\/span>.address = address;<\/span>
    }<\/span>
<\/span>
    public<\/span> int<\/span> getSex<\/span>()<\/span> <\/span>{<\/span>
        return<\/span> sex;<\/span>
    }<\/span>
<\/span>
    public<\/span> void<\/span> setSex<\/span>(int<\/span> sex)<\/span> <\/span>{<\/span>
        this<\/span>.sex = sex;<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> String toString<\/span>()<\/span> <\/span>{<\/span>
        return<\/span> \"User{\"<\/span> +<\/span>
                \"id=\"<\/span> + id +<\/span>
                \", name='\"<\/span> + name + '''<\/span> +<\/span>
                \", address='\"<\/span> + address + '''<\/span> +<\/span>
                \", sex=\"<\/span> + sex +<\/span>
                '}'<\/span>;<\/span>
    }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>\u5b9e\u73b0flink \u81ea\u5b9a\u4e49mysql source<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
51<\/span>
52<\/span>
53<\/span>
54<\/span>
55<\/span>
56<\/span>
57<\/span>
58<\/span>
59<\/span>
60<\/span>
61<\/span>
62<\/span>
63<\/span>
64<\/span>
65<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.java.source;<\/span>
<\/span>
import<\/span> org.apache.flink.configuration.Configuration;<\/span>
import<\/span> org.apache.flink.streaming.api.functions.source.RichSourceFunction;<\/span>
import<\/span> xin.studytime.java.model.User;<\/span>
<\/span>
import<\/span> java.sql.DriverManager;<\/span>
import<\/span> java.sql.PreparedStatement;<\/span>
import<\/span> java.sql.Connection;<\/span>
import<\/span> java.sql.ResultSet;<\/span>
<\/span>
public<\/span> class<\/span> UserSourceFromMysql<\/span> extends<\/span> RichSourceFunction<\/span><User<\/span>> <\/span>{<\/span>
    PreparedStatement preparedStatement;<\/span>
    private<\/span> Connection connection;<\/span>
<\/span>
    public<\/span> UserSourceFromMysql<\/span>()<\/span> <\/span>{<\/span>
        super<\/span>();<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> open<\/span>(Configuration parameters)<\/span> throws<\/span> Exception <\/span>{<\/span>
        super<\/span>.open( parameters );<\/span>
        String driver = \"com.mysql.jdbc.Driver\"<\/span>;<\/span>
        String url = \"jdbc:mysql:\/\/node02:3306\/flink\"<\/span>;<\/span>
        String username = \"root\"<\/span>;<\/span>
        String password = \"!Qaz123456\"<\/span>;<\/span>
<\/span>
        Class.forName( driver );<\/span>
        connection = DriverManager.getConnection( url, username, password );<\/span>
        String sql = \"select * from user;\"<\/span>;<\/span>
        preparedStatement = connection.prepareStatement( sql );<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> close<\/span>()<\/span> throws<\/span> Exception <\/span>{<\/span>
        super<\/span>.close();<\/span>
        if<\/span> (connection != null<\/span>) {<\/span>
            connection.close();<\/span>
        }<\/span>
        if<\/span> (preparedStatement != null<\/span>) {<\/span>
            preparedStatement.close();<\/span>
        }<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> run<\/span>(SourceContext<User> sourceContext)<\/span> throws<\/span> Exception <\/span>{<\/span>
        try<\/span> {<\/span>
            ResultSet resultSet = preparedStatement.executeQuery();<\/span>
            while<\/span> (resultSet.next()) {<\/span>
                User user = new<\/span> User( resultSet.getInt( \"id\"<\/span> ),<\/span>
                        resultSet.getString( \"name\"<\/span> ).trim(),<\/span>
                        resultSet.getString( \"address\"<\/span> ).trim(),<\/span>
                        resultSet.getInt( \"sex\"<\/span> ) );<\/span>
                sourceContext.collect( user );<\/span>
            }<\/span>
        } catch<\/span> (Exception e) {<\/span>
            e.printStackTrace();<\/span>
        }<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> cancel<\/span>()<\/span> <\/span>{<\/span>
<\/span>
    }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>flink \u4e3b\u7a0b\u5e8f\u5f15\u5165source\uff0c\u6d4b\u8bd5\u6570\u636e\u8bfb\u53d6<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.java;<\/span>
<\/span>
import<\/span> org.apache.flink.streaming.api.datastream.DataStreamSource;<\/span>
import<\/span> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;<\/span>
import<\/span> xin.studytime.java.model.User;<\/span>
<\/span>
public<\/span> class<\/span> UserSourceFromMysql<\/span> <\/span>{<\/span>
    public<\/span> static<\/span> void<\/span> main<\/span>(String[] args)<\/span> throws<\/span> Exception <\/span>{<\/span>
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();<\/span>
<\/span>
        DataStreamSource<User> userDataStreamSource = env.addSource( new<\/span> xin.studytime.java.source.UserSourceFromMysql() );<\/span>
<\/span>
        userDataStreamSource.print();<\/span>
<\/span>
        env.execute();<\/span>
    }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u8fd0\u884c\u7ed3\u679c\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
<\/pre>\n<\/td>\n
\n
\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/bin\/java -agentlib:jdwp=transport=dt_socket,address=127.0.0.1:58791,suspend=y,server=n -javaagent:\/Users\/baihe\/Library\/Caches\/JetBrains\/IntelliJIdea2020.1\/captureAgent\/debugger-agent.jar -Dfile.encoding=UTF-8 -classpath \/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/charsets.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/deploy.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/cldrdata.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/dnsns.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/jaccess.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/jfxrt.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/localedata.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/nashorn.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/sunec.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/sunjce_provider.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/sunpkcs11.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/ext\/zipfs.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/javaws.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/jce.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/jfr.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/jfxswt.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/jsse.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/management-agent.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/plugin.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/resources.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/jre\/lib\/rt.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/ant-javafx.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/dt.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/javafx-mx.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/jconsole.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/packager.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/sa-jdi.jar:\/Library\/Java\/JavaVirtualMachines\/jdk1.8.0_251.jdk\/Contents\/Home\/lib\/tools.jar:\/Users\/baihe\/Code\/javaCode\/flink-project\/target\/classes:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-library.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-xml_2.11-1.0.4.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-parser-combinators_2.11-1.0.4.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-actors-migration_2.11-1.1.0.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-actors-2.11.0.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-swing_2.11-1.0.2.jar:\/Users\/baihe\/Code\/tools\/scala-2.11.8\/lib\/scala-reflect.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-streaming-java_2.11\/1.9.0\/flink-streaming-java_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-core\/1.9.0\/flink-core-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-annotations\/1.9.0\/flink-annotations-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-metrics-core\/1.9.0\/flink-metrics-core-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-shaded-asm-6\/6.2.1-7.0\/flink-shaded-asm-6-6.2.1-7.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/commons\/commons-lang3\/3.3.2\/commons-lang3-3.3.2.jar:\/Users\/baihe\/.m2\/repository\/com\/esotericsoftware\/kryo\/kryo\/2.24.0\/kryo-2.24.0.jar:\/Users\/baihe\/.m2\/repository\/com\/esotericsoftware\/minlog\/minlog\/1.2\/minlog-1.2.jar:\/Users\/baihe\/.m2\/repository\/org\/objenesis\/objenesis\/2.1\/objenesis-2.1.jar:\/Users\/baihe\/.m2\/repository\/commons-collections\/commons-collections\/3.2.2\/commons-collections-3.2.2.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/commons\/commons-compress\/1.18\/commons-compress-1.18.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-runtime_2.11\/1.9.0\/flink-runtime_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-java\/1.9.0\/flink-java-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-queryable-state-client-java\/1.9.0\/flink-queryable-state-client-java-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-hadoop-fs\/1.9.0\/flink-hadoop-fs-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/commons-io\/commons-io\/2.4\/commons-io-2.4.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-shaded-netty\/4.1.32.Final-7.0\/flink-shaded-netty-4.1.32.Final-7.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-shaded-jackson\/2.9.8-7.0\/flink-shaded-jackson-2.9.8-7.0.jar:\/Users\/baihe\/.m2\/repository\/commons-cli\/commons-cli\/1.3.1\/commons-cli-1.3.1.jar:\/Users\/baihe\/.m2\/repository\/org\/javassist\/javassist\/3.19.0-GA\/javassist-3.19.0-GA.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/akka\/akka-actor_2.11\/2.5.21\/akka-actor_2.11-2.5.21.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/config\/1.3.3\/config-1.3.3.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/modules\/scala-java8-compat_2.11\/0.7.0\/scala-java8-compat_2.11-0.7.0.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/akka\/akka-stream_2.11\/2.5.21\/akka-stream_2.11-2.5.21.jar:\/Users\/baihe\/.m2\/repository\/org\/reactivestreams\/reactive-streams\/1.0.2\/reactive-streams-1.0.2.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/ssl-config-core_2.11\/0.3.7\/ssl-config-core_2.11-0.3.7.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/akka\/akka-protobuf_2.11\/2.5.21\/akka-protobuf_2.11-2.5.21.jar:\/Users\/baihe\/.m2\/repository\/com\/typesafe\/akka\/akka-slf4j_2.11\/2.5.21\/akka-slf4j_2.11-2.5.21.jar:\/Users\/baihe\/.m2\/repository\/org\/clapper\/grizzled-slf4j_2.11\/1.3.2\/grizzled-slf4j_2.11-1.3.2.jar:\/Users\/baihe\/.m2\/repository\/com\/github\/scopt\/scopt_2.11\/3.5.0\/scopt_2.11-3.5.0.jar:\/Users\/baihe\/.m2\/repository\/org\/xerial\/snappy\/snappy-java\/1.1.4\/snappy-java-1.1.4.jar:\/Users\/baihe\/.m2\/repository\/com\/twitter\/chill_2.11\/0.7.6\/chill_2.11-0.7.6.jar:\/Users\/baihe\/.m2\/repository\/com\/twitter\/chill-java\/0.7.6\/chill-java-0.7.6.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-clients_2.11\/1.9.0\/flink-clients_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-optimizer_2.11\/1.9.0\/flink-optimizer_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-shaded-guava\/18.0-7.0\/flink-shaded-guava-18.0-7.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/commons\/commons-math3\/3.5\/commons-math3-3.5.jar:\/Users\/baihe\/.m2\/repository\/org\/slf4j\/slf4j-api\/1.7.15\/slf4j-api-1.7.15.jar:\/Users\/baihe\/.m2\/repository\/com\/google\/code\/findbugs\/jsr305\/1.3.9\/jsr305-1.3.9.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/force-shading\/1.9.0\/force-shading-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-streaming-scala_2.11\/1.9.0\/flink-streaming-scala_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/apache\/flink\/flink-scala_2.11\/1.9.0\/flink-scala_2.11-1.9.0.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/scala-reflect\/2.11.12\/scala-reflect-2.11.12.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/scala-library\/2.11.12\/scala-library-2.11.12.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/scala-compiler\/2.11.12\/scala-compiler-2.11.12.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/modules\/scala-xml_2.11\/1.0.5\/scala-xml_2.11-1.0.5.jar:\/Users\/baihe\/.m2\/repository\/org\/scala-lang\/modules\/scala-parser-combinators_2.11\/1.0.4\/scala-parser-combinators_2.11-1.0.4.jar:\/Users\/baihe\/.m2\/repository\/org\/slf4j\/slf4j-nop\/1.7.2\/slf4j-nop-1.7.2.jar:\/Users\/baihe\/.m2\/repository\/mysql\/mysql-connector-java\/5.1.35\/mysql-connector-java-5.1.35.jar:\/Applications\/IntelliJ IDEA.app\/Contents\/lib\/idea_rt.jar xin.studytime.java.UserSourceFromMysql<\/span>
Connected to the target VM, address: '127.0.0.1:58791', transport: 'socket'<\/span>
6><\/span> User{id=2, name='xiaoqiang'<\/span>, address='GuangZhou'<\/span>, sex=2}<\/span><\/span>
5><\/span> User{id=1, name='xiaoming'<\/span>, address='Beijing'<\/span>, sex=1}<\/span><\/span>
2><\/span> User{id=6, name='lisi'<\/span>, address='Hunan'<\/span>, sex=1}<\/span><\/span>
1><\/span> User{id=5, name='zhangsan'<\/span>, address='Beijing'<\/span>, sex=1}<\/span><\/span>
7><\/span> User{id=3, name='xiaohua'<\/span>, address='Hubei'<\/span>, sex=0}<\/span><\/span>
8><\/span> User{id=4, name='xiaoli'<\/span>, address='GuangZhou'<\/span>, sex=1}<\/span><\/span>
Disconnected from the target VM, address: '127.0.0.1:58791', transport: 'socket'<\/span>
<\/span>
Process finished with exit code 0<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>\u5b9e\u73b0flink \u81ea\u5b9a\u4e49mysql sink<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.java.sink;<\/span>
<\/span>
import<\/span> org.apache.flink.configuration.Configuration;<\/span>
import<\/span> org.apache.flink.streaming.api.functions.sink.RichSinkFunction;<\/span>
import<\/span> xin.studytime.java.model.User;<\/span>
<\/span>
import<\/span> java.sql.Connection;<\/span>
import<\/span> java.sql.DriverManager;<\/span>
import<\/span> java.sql.PreparedStatement;<\/span>
<\/span>
public<\/span> class<\/span> UserSinkToMysql<\/span> extends<\/span> RichSinkFunction<\/span><User<\/span>> <\/span>{<\/span>
    private<\/span> Connection connection = null<\/span>;<\/span>
    private<\/span> PreparedStatement preparedStatement = null<\/span>;<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> open<\/span>(Configuration parameters)<\/span> throws<\/span> Exception <\/span>{<\/span>
        super<\/span>.open( parameters );<\/span>
        String driver = \"com.mysql.jdbc.Driver\"<\/span>;<\/span>
        String url = \"jdbc:mysql:\/\/node02:3306\/flink\"<\/span>;<\/span>
        String username = \"root\"<\/span>;<\/span>
        String password = \"!Qaz123456\"<\/span>;<\/span>
        Class.forName( driver );<\/span>
        connection = DriverManager.getConnection( url, username, password );<\/span>
        String sql = \"insert into user(name,address,sex)values(?,?,?);\"<\/span>;<\/span>
        preparedStatement = connection.prepareStatement( sql );<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> close<\/span>()<\/span> throws<\/span> Exception <\/span>{<\/span>
        super<\/span>.close();<\/span>
        if<\/span> (connection != null<\/span>) {<\/span>
            connection.close();<\/span>
        }<\/span>
        if<\/span> (preparedStatement != null<\/span>) {<\/span>
            preparedStatement.close();<\/span>
        }<\/span>
    }<\/span>
<\/span>
    @Override<\/span><\/span>
    public<\/span> void<\/span> invoke<\/span>(User value, Context context)<\/span> throws<\/span> Exception <\/span>{<\/span>
        try<\/span> {<\/span>
            preparedStatement.setString( 1<\/span>, value.getName() );<\/span>
            preparedStatement.setString( 2<\/span>, value.getAddress() );<\/span>
            preparedStatement.setInt( 3<\/span>, value.getSex() );<\/span>
            preparedStatement.executeUpdate();<\/span>
        } catch<\/span> (Exception e) {<\/span>
            e.printStackTrace();<\/span>
        }<\/span>
    }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>flink \u4e3b\u7a0b\u5e8f\u5f15\u5165sink\uff0c\u6d4b\u8bd5\u6570\u636e\u5199\u5165mysql<\/h2>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.java;<\/span>
<\/span>
import<\/span> org.apache.flink.streaming.api.datastream.DataStream;<\/span>
import<\/span> org.apache.flink.streaming.api.datastream.DataStreamSource;<\/span>
import<\/span> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;<\/span>
import<\/span> xin.studytime.java.model.User;<\/span>
<\/span>
public<\/span> class<\/span> UserSinkToMysql<\/span> <\/span>{<\/span>
    public<\/span> static<\/span> void<\/span> main<\/span>(String[] args)<\/span> throws<\/span> Exception<\/span>{<\/span>
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();<\/span>
        DataStream<User> users = env.fromElements( new<\/span> User( \"zhangsan\"<\/span>, \"Beijing\"<\/span>, 1<\/span> ), new<\/span> User( \"lisi\"<\/span>, \"Hunan\"<\/span>, 1<\/span> ) );<\/span>
<\/span>
        users.addSink( new<\/span> xin.studytime.java.sink.UserSinkToMysql() );<\/span>
<\/span>
<\/span>
        env.execute();<\/span>
    }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
\u8fd0\u884c\u7ed3\u679c\uff1a<\/p>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
<\/pre>\n<\/td>\n
\n
+----+-----------+-----------+-----+---------------------+---------------------+<\/span>
| id | name      | address   | sex | create_time         | update_time         |<\/span>
+----+-----------+-----------+-----+---------------------+---------------------+<\/span>
| 1  | xiaoming  | Beijing   | 1   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 2  | xiaoqiang | GuangZhou | 2   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 3  | xiaohua   | Hubei     | 0   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 4  | xiaoli    | GuangZhou | 1   | 2020-08-30 15:27:05 | 2020-08-30 15:27:05 |<\/span>
| 5  | zhangsan  | Beijing   | 1   | 2020-08-30 16:11:28 | 2020-08-30 16:11:28 |<\/span>
| 6  | lisi      | Hunan     | 1   | 2020-08-30 16:11:28 | 2020-08-30 16:11:28 |<\/span>
+----+-----------+-----------+-----+---------------------+---------------------+<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n","protected":false},"excerpt":{"rendered":"
\u6570\u636e\u5e93\u8868\u51c6\u5907 1234567891…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[],"tags":[],"_links":{"self":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/148048"}],"collection":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=148048"}],"version-history":[{"count":0,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/148048\/revisions"}],"wp:attachment":[{"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=148048"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=148048"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=148048"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}