标签：jdbc java 读取数据 wordName wordcount public mysql id String

本文的基础环境可以参考flink 1.10.1 java版本wordcount演示 (nc + socket)，在此基础上实现通过jdbc从mysql读取数据。

1. 添加依赖

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.18</version>
</dependency>

由于测试环境是mysql8，所以这里选择的是mysql8.0.18版本的驱动。

 2. 实体类

package com.demo.source;


public class WordCount {
    private Long id;
    private String wordName;
    private Long wordcount;

    public WordCount() {
    }

    public WordCount(Long id, String wordName, Long wordcount) {
        this.id = id;
        this.wordName = wordName;
        this.wordcount = wordcount;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getWordName() {
        return wordName;
    }

    public void setWordName(String wordName) {
        this.wordName = wordName;
    }

    public Long getWordcount() {
        return wordcount;
    }

    public void setWordcount(Long wordcount) {
        this.wordcount = wordcount;
    }

    @Override
    public String toString() {
        return "WordCount{" +
                "id=" + id +
                ", wordName='" + wordName + '\'' +
                ", wordcount=" + wordcount +
                '}';
    }
}

实体类定义了一个无惨构造和一个全参数构造函数，重写了toString方法。

3. 定义mysql source

package com.demo.source;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 *
 */
public class MySqlSource extends RichSourceFunction<WordCount> {

    private PreparedStatement ps=null;
    private Connection connection=null;
    String driver = "com.mysql.jdbc.Driver";
    String url = "jdbc:mysql://localhost:3306/test?serverTimezone=GMT%2b8";
    String username  = "username";
    String password = "password";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        connection = getConnection();
        String sql = "select id, word_name, word_count from tbl_flink_wordcount;";
        //获取执行语句
        ps = connection.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<WordCount> sourceContext) throws Exception {

        ResultSet resultSet = ps.executeQuery();
        while (resultSet.next()){
            WordCount student = new WordCount(
                    resultSet.getLong("id"),
                    resultSet.getString("word_name").trim(),
                    resultSet.getLong("word_count"));

            sourceContext.collect(student);//发送结果
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();

        if(connection != null){
            connection.close();
        }
        if (ps != null){
            ps.close();
        }
    }

    //获取mysql连接配置
    public Connection getConnection(){
        try {
            // 加载驱动
            // Class.forName(driver);
            //创建连接
            connection = DriverManager.getConnection(url,username,password);
        } catch (Exception e) {
            System.out.println("********mysql get connection occur exception, msg = "+e.getMessage());
            e.printStackTrace();
        }
        return  connection;
    }

}

采用MySQL8驱动连接mysql8时，需要指定时区。

如果数据量大，还可以考虑分页读取等方式进行处理。

由于数据读取过程和数据处理过程是同时进行的，所以读取过程可以持续读取，也可以延时读取，一直处于读取过程中，也不会影响数据处理过程。

4. 测试代码

package com.demo.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *
 */
public class FlinkMySqlSourceDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WordCount> dataStream = env.addSource(new MySqlSource());
        
        // 指定并行度为1
        dataStream.print().setParallelism(1);

        env.execute("Flink Mysql Source");

    }
}

5. 启动程序执行测试

可以看到从mysql读取的数据。

 6. 参考表定义

CREATE TABLE `tbl_flink_wordcount` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `word_name` varchar(64) DEFAULT NULL,
  `word_count` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8;

标签：jdbc,java,读取数据,wordName,wordcount,public,mysql,id,String
来源： https://blog.csdn.net/liaomingwu/article/details/122835030

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。