JDBC 커넥터

JDBC 커넥터 (JDBC Connector)

이 커넥터는 JDBC 데이터베이스에 데이터를 쓰는 싱크를 제공해요. 사용하려면 JDBC 드라이버와 함께 다음 의존성을 프로젝트에 추가해요.

출처: 문서

본문

Flink 2.3 버전에는 아직 사용할 수 있는 커넥터가 없어요. 스트리밍 커넥터는 현재 바이너리 배포의 일부가 아닙니다. 클러스터 실행을 위해 함께 연결하는 방법은 여기를 참고해요. 지정된 데이터베이스에 연결하려면 드라이버 의존성도 필요해요. 해당 드라이버를 추가하는 방법은 데이터베이스 문서를 참고해요.

JdbcSink.sink

JDBC 싱크는 at-least-once 보장을 제공해요. 하지만 효과적으로 upsert SQL 문이나 멱등 SQL 업데이트를 만들어 exactly-once를 달성할 수 있어요. 구성은 다음과 같아요(JdbcSink javadoc 참고):

Java:

JdbcSink.sink(
sqlDmlStatement, // mandatory
jdbcStatementBuilder, // mandatory
jdbcExecutionOptions, // optional
jdbcConnectionOptions // mandatory
);

Python:

JdbcSink.sink(
sql_dml_statement, # mandatory
type_info, # mandatory
jdbc_connection_options, # mandatory
jdbc_execution_options # optional
)

SQL DML 문과 JDBC 문 빌더 (SQL DML statement and JDBC statement builder)

싱크는 사용자가 제공한 SQL 문자열에서 하나의 JDBC prepared statement를 만들어요. 예:

INSERT INTO some_table field1, field2 values (?, ?)

그런 다음 스트림의 각 값으로 그 prepared statement를 갱신하는 사용자 제공 함수를 반복 호출해요. 예:

(preparedStatement, someRecord) -> { ... update here the preparedStatement with values from someRecord ... }

JDBC 실행 옵션 (JDBC execution options)

SQL DML 문은 배치로 실행되며 다음 인스턴스로 선택적으로 구성할 수 있어요(JdbcExecutionOptions javadoc 참고):

Java:

JdbcExecutionOptions.builder()
.withBatchIntervalMs(200) // optional: default = 0, meaning no time-based execution is done
.withBatchSize(1000) // optional: default = 5000 values
.withMaxRetries(5) // optional: default = 3
.build();

Python:

JdbcExecutionOptions.builder() \
.with_batch_interval_ms(2000) \
.with_batch_size(100) \
.with_max_retries(5) \
.build()

다음 조건 중 하나가 참이 되는 즉시 JDBC 배치가 실행돼요:

• 설정된 배치 간격 시간이 경과했을 때
• 최대 배치 크기에 도달했을 때
• Flink 체크포인트가 시작되었을 때

JDBC 연결 파라미터 (JDBC connection parameters)

데이터베이스 연결은 JdbcConnectionOptions 인스턴스로 구성돼요. 자세한 내용은 JdbcConnectionOptions javadoc을 참고해요.

완전한 예제 (Full example)

Java:

public class JdbcSinkExample {

static class Book {
public Book(Long id, String title, String authors, Integer year) {
this.id = id;
this.title = title;
this.authors = authors;
this.year = year;
}
final Long id;
final String title;
final String authors;
final Integer year;
}

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

env.fromElements(
new Book(101L, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
new Book(102L, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
new Book(103L, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
new Book(104L, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017)
).addSink(
JdbcSink.sink(
"insert into books (id, title, authors, year) values (?, ?, ?, ?)",
(statement, book) -> {
statement.setLong(1, book.id);
statement.setString(2, book.title);
statement.setString(3, book.authors);
statement.setInt(4, book.year);
},
JdbcExecutionOptions.builder()
.withBatchSize(1000)
.withBatchIntervalMs(200)
.withMaxRetries(5)
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
.withDriverName("org.postgresql.Driver")
.withUsername("someUser")
.withPassword("somePassword")
.build()
));

env.execute();
}
}

Python:

env = StreamExecutionEnvironment.get_execution_environment()
type_info = Types.ROW([Types.INT(), Types.STRING(), Types.STRING(), Types.INT()])
env.from_collection(
[(101, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
(102, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
(103, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
(104, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017)
], type_info=type_info) \
.add_sink(
JdbcSink.sink(
"insert into books (id, title, authors, year) values (?, ?, ?, ?)",
type_info,
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.with_url('jdbc:postgresql://dbhost:5432/postgresdb')
.with_driver_name('org.postgresql.Driver')
.with_user_name('someUser')
.with_password('somePassword')
.build(),
JdbcExecutionOptions.builder()
.with_batch_interval_ms(1000)
.with_batch_size(200)
.with_max_retries(5)
.build()
))

env.execute()

JdbcSink.exactlyOnceSink

1.13부터 Flink JDBC 싱크는 exactly-once 모드를 지원해요. 구현은 JDBC 드라이버의 XA 표준 지원에 의존해요. 데이터베이스가 XA를 지원하면 대부분의 드라이버가 XA를 지원해요(드라이버는 보통 같기 때문). 사용하려면 위와 같이 exactlyOnceSink() 메서드로 싱크를 만들고 추가로 다음을 제공해요:

• exactly-once options
• execution options
• XA DataSource 공급자

예를 들어:

Java:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.fromElements(...)
.addSink(JdbcSink.exactlyOnceSink(
"insert into books (id, title, author, price, qty) values (?,?,?,?,?)",
(ps, t) -> {
ps.setInt(1, t.id);
ps.setString(2, t.title);
ps.setString(3, t.author);
ps.setDouble(4, t.price);
ps.setInt(5, t.qty);
},
JdbcExecutionOptions.builder()
.withMaxRetries(0)
.build(),
JdbcExactlyOnceOptions.defaults(),
() -> {
// create a driver-specific XA DataSource
// The following example is for derby
EmbeddedXADataSource ds = new EmbeddedXADataSource();
ds.setDatabaseName("my_db");
return ds;
});
env.execute();

Python:

Still not supported in Python API.

참고: 일부 데이터베이스(예: PostgreSQL, MySQL)는 연결당 단일 XA 트랜잭션만 허용해요. 이런 경우 JdbcExactlyOnceOptions를 구성하려면 다음 API를 사용해요:

Java:

JdbcExactlyOnceOptions.builder()
.withTransactionPerConnection(true)
.build();

Python:

Still not supported in Python API.

이렇게 하면 Flink가 모든 XA 트랜잭션에 별도 연결을 사용해요. 연결 한도 조정이 필요할 수 있어요. PostgreSQL과 MySQL의 경우 max_connections을 늘리면 됩니다. 또한 일부 데이터베이스에서 XA를 활성화하거나 구성해야 해요. PostgreSQL에서는 max_prepared_transactions를 0보다 큰 값으로 설정해야 해요. MySQL v8+에서는 Flink DB 사용자에게 XA_RECOVER_ADMIN을 부여해야 해요.

주의: 현재 JdbcSink.exactlyOnceSink는 JdbcExecutionOptions.maxRetries == 0일 때만 정확히 once 의미론을 보장할 수 있어요. 그렇지 않으면 중복 결과가 만들어질 수 있어요.

XADataSource 예제

PostgreSQL XADataSource 예제:

Java:

PGXADataSource xaDataSource = new org.postgresql.xa.PGXADataSource();
xaDataSource.setUrl("jdbc:postgresql://localhost:5432/postgres");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);

Python:

Still not supported in Python API.

MySQL XADataSource 예제:

Java:

MysqlXADataSource xaDataSource = new com.mysql.cj.jdbc.MysqlXADataSource();
xaDataSource.setUrl("jdbc:mysql://localhost:3306/");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);

Python:

Still not supported in Python API.

Oracle XADataSource 예제:

Java:

OracleXADataSource xaDataSource = new oracle.jdbc.xa.OracleXADataSource();
xaDataSource.setURL("jdbc:oracle:oci8:@");
xaDataSource.setUser("scott");
xaDataSource.setPassword("tiger");

Python:

Still not supported in Python API.

Oracle 연결 풀링도 고려해주세요. 자세한 내용은 JdbcXaSinkFunction 문서를 참고해요.

더 알아보기 (Learn more)

• DataStream 커넥터 개요
• Kafka 커넥터
• Kinesis 커넥터