クラウドスターデータ---Apache Flink実戦シリーズ(逸品版):Flinkストリーム処理API詳細とプログラミング実戦010-DataStreamとMySqlカスタムsinkとsource(Java版)002
5624 ワード
三、カスタムソース
1.カスタムソース
2.sourceテストプログラム
3.sourceテスト効果
1.カスタムソース
package code.book.stream.customsinkandsource.jdbc.java;
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 StudentSourceFromMysql extends RichSourceFunction<Student> {
PreparedStatement ps;
private Connection connection;
/**
* 、open() , invoke 。
*/
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
String driver = "com.mysql.jdbc.Driver";
String url = "jdbc:mysql://qingcheng11:3306/flinktest";
String username = "root";
String password = "qingcheng";
//1.
Class.forName(driver);
//2.
connection = DriverManager.getConnection(url, username, password);
//3.
String sql = "select stuid,stuname,stuaddr,stusex from Student;";
ps = connection.prepareStatement(sql);
}
/**
* 、DataStream run()
*/
@Override
public void run(SourceContext sourceContext) throws Exception {
try {
//4. ,
ResultSet resultSet = ps.executeQuery();
while (resultSet.next()) {
Student student = new Student(
resultSet.getInt("stuid"),
resultSet.getString("stuname").trim(),
resultSet.getString("stuaddr").trim(),
resultSet.getString("stusex").trim());
sourceContext.collect(student);
}
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public void cancel() {
}
/**
* 、 ,
*/
@Override
public void close() throws Exception {
//5.
super.close();
if (connection != null) {
connection.close();
}
if (ps != null) {
ps.close();
}
}
}
2.sourceテストプログラム
package code.book.stream.customsinkandsource.jdbc.java;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class StudentSourceFromMysqlTest {
public static void main(String[] args) throws Exception {
//1.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2. source
DataStream students=env.addSource(new StudentSourceFromMysql());
//3.
students.print();
//4.
env.execute();
}
}
3.sourceテスト効果
mysql 。