flink 读写 mysql

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

package com.run;

import java.sql.DriverManager;
import java.sql.ResultSet;

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

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class Flink2JdbcReader extends
		RichSourceFunction<Tuple10<String, String, String, String, String, String, String, String, String, String>> {
	private static final long serialVersionUID = 3334654984018091675L;

	private Connection connect = null;
	private PreparedStatement ps = null;

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
	 * apache.flink.configuration.Configuration) to use open database connect
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		Class.forName("com.mysql.jdbc.Driver");
		connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.21.11:3306", "root", "flink");
		ps = (PreparedStatement) connect
				.prepareStatement("select col1,col2,col3,col4,col5,col6,col7,col8,col9,col10 from flink.test_tb");
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.apache.flink.streaming.api.functions.source.SourceFunction#run(org.
	 * apache.flink.streaming.api.functions.source.SourceFunction.SourceContext)
	 * to use excuted sql and return result
	 */
	@Override
	public void run(
			SourceContext<Tuple10<String, String, String, String, String, String, String, String, String, String>> collect)
			throws Exception {
		ResultSet resultSet = ps.executeQuery();
		while (resultSet.next()) {
			Tuple10<String, String, String, String, String, String, String, String, String, String> tuple = new Tuple10<String, String, String, String, String, String, String, String, String, String>();
			tuple.setFields(resultSet.getString(1), resultSet.getString(2), resultSet.getString(3),
					resultSet.getString(4), resultSet.getString(5), resultSet.getString(6), resultSet.getString(7),
					resultSet.getString(8), resultSet.getString(9), resultSet.getString(10));
			collect.collect(tuple);
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.apache.flink.streaming.api.functions.source.SourceFunction#cancel()
	 * colse database connect
	 */
	@Override
	public void cancel() {
		try {
			super.close();
			if (connect != null) {
				connect.close();
			}
			if (ps != null) {
				ps.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

package com.run;

import java.sql.DriverManager;

import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class Flink2JdbcWriter extends
		RichSinkFunction<Tuple10<String, String, String, String, String, String, String, String, String, String>> {
	private static final long serialVersionUID = -8930276689109741501L;

	private Connection connect = null;
	private PreparedStatement ps = null;

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
	 * apache.flink.configuration.Configuration) get database connect
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		Class.forName("com.mysql.jdbc.Driver");
		connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.21.11:3306", "root", "flink");
		ps = (PreparedStatement) connect.prepareStatement("insert into flink.test_tb1 values (?,?,?,?,?,?,?,?,?,?)");
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.apache.flink.streaming.api.functions.sink.SinkFunction#invoke(java.
	 * lang.Object,
	 * org.apache.flink.streaming.api.functions.sink.SinkFunction.Context) read
	 * data from flink DataSet to database
	 */
	@Override
	public void invoke(Tuple10<String, String, String, String, String, String, String, String, String, String> value,
			Context context) throws Exception {
		ps.setString(1, value.f0);
		ps.setString(2, value.f1);
		ps.setString(3, value.f2);
		ps.setString(4, value.f3);
		ps.setString(5, value.f4);
		ps.setString(6, value.f5);
		ps.setString(7, value.f6);
		ps.setString(8, value.f7);
		ps.setString(9, value.f8);
		ps.setString(10, value.f9);
		ps.executeUpdate();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.apache.flink.api.common.functions.AbstractRichFunction#close()
	 * close database connect
	 */
	@Override
	public void close() throws Exception {
		try {
			super.close();
			if (connect != null) {
				connect.close();
			}
			if (ps != null) {
				ps.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

package com.run;

import java.util.Date;

import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkReadDbWriterDb {
	public static void main(String[] args) throws Exception {。
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStream<Tuple10<String, String, String, String, String, String, String, String, String, String>> dataStream = env
				.addSource(new Flink2JdbcReader());
		
		// tranfomat
		
		dataStream.addSink(new Flink2JdbcWriter());
		env.execute("Flink cost DB data to write Database");
		
	}
}