flink Sink簡(jiǎn)介
- flink 中有兩個(gè)重要的概念，Source 和 Sink ，Source 決定了我們的數(shù)據(jù)從哪里產(chǎn)生，而 Sink 決定了數(shù)據(jù)將要去到什么地方。
- flink 自帶有豐富的 Sink，比如：kafka、csv 文件、ES、Socket 等等。
- 當(dāng)我們想要使用當(dāng)前并未實(shí)現(xiàn)的 Sink 函數(shù)時(shí)，可以進(jìn)行自定義。
自定義 Sink 函數(shù)
- 這里主要自定義寫入 kudu 的 kuduSink。
- 自定義sink需要我們實(shí)現(xiàn) SinkFunction，或者繼承 RichSinkFunction，下面我們閱讀源碼來對(duì)其進(jìn)行比較：
SinkFunction 函數(shù)，這是一個(gè)接口。
package org.apache.flink.streaming.api.functions.sink;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.Function;
import java.io.Serializable;
/**
* Interface for implementing user defined sink functionality.
*
* @param <IN> Input type parameter.
*/
@Public
public interface SinkFunction<IN> extends Function, Serializable {
/**
* @deprecated Use {@link #invoke(Object, Context)}.
*/
@Deprecated
default void invoke(IN value) throws Exception {}
/**
* Writes the given value to the sink. This function is called for every record.
*
* <p>You have to override this method when implementing a {@code SinkFunction}, this is a
* {@code default} method for backward compatibility with the old-style method only.
*
* @param value The input record.
* @param context Additional context about the input record.
*
* @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
* to fail and may trigger recovery.
*/
default void invoke(IN value, Context context) throws Exception {
invoke(value);
}
/**
* Context that {@link SinkFunction SinkFunctions } can use for getting additional data about
* an input record.
*
* <p>The context is only valid for the duration of a
* {@link SinkFunction#invoke(Object, Context)} call. Do not store the context and use
* afterwards!
*
* @param <T> The type of elements accepted by the sink.
*/
@Public // Interface might be extended in the future with additional methods.
interface Context<T> {
/** Returns the current processing time. */
long currentProcessingTime();
/** Returns the current event-time watermark. */
long currentWatermark();
/**
* Returns the timestamp of the current input record or {@code null} if the element does not
* have an assigned timestamp.
*/
Long timestamp();
}
}
RichSinkFunction 函數(shù)，這個(gè)是一個(gè)抽象類。
package org.apache.flink.streaming.api.functions.sink;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.AbstractRichFunction;
/**
* A {@link org.apache.flink.api.common.functions.RichFunction} version of {@link SinkFunction}.
*/
@Public
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {
private static final long serialVersionUID = 1L;
}
由源碼可以看到，RichSinkFunction 抽象類繼承了 SinkFunction 接口，在使用過程中會(huì)更加靈活。通常情況下，在自定義 Sink 函數(shù)時(shí)，是繼承 RichSinkFunction 來實(shí)現(xiàn)。
KuduSink 函數(shù), 繼承了 RichSinkFunction，重寫了 open、close 和 invoke 方法，在 open 中進(jìn)行 kudu 相關(guān)配置的初始化，在 invoke 中進(jìn)行數(shù)據(jù)寫入的相關(guān)操作，最后在 close 中關(guān)掉所有的開關(guān)。
package test;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.apache.log4j.Logger;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.Map;
public class SinkKudu extends RichSinkFunction<Map<String, Object>> {
private final static Logger logger = Logger.getLogger(SinkKudu.class);
private KuduClient client;
private KuduTable table;
private String kuduMaster;
private String tableName;
private Schema schema;
private KuduSession kuduSession;
private ByteArrayOutputStream out;
private ObjectOutputStream os;
public SinkKudu(String kuduMaster, String tableName) {
this.kuduMaster = kuduMaster;
this.tableName = tableName;
}
@Override
public void open(Configuration parameters) throws Exception {
out = new ByteArrayOutputStream();
os = new ObjectOutputStream(out);
client = new KuduClient.KuduClientBuilder(kuduMaster).build();
table = client.openTable(tableName);
schema = table.getSchema();
kuduSession = client.newSession();
kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
}
@Override
public void invoke(Map<String, Object> map) {
if (map == null) {
return;
}
try {
int columnCount = schema.getColumnCount();
Insert insert = table.newInsert();
PartialRow row = insert.getRow();
for (int i = 0; i < columnCount; i++) {
Object value = map.get(schema.getColumnByIndex(i).getName());
insertData(row, schema.getColumnByIndex(i).getType(), schema.getColumnByIndex(i).getName(), value);
}
OperationResponse response = kuduSession.apply(insert);
if (response != null) {
logger.error(response.getRowError().toString());
}
} catch (Exception e) {
logger.error(e);
}
}
@Override
public void close() throws Exception {
try {
kuduSession.close();
client.close();
os.close();
out.close();
} catch (Exception e) {
logger.error(e);
}
}
// 插入數(shù)據(jù)
private void insertData(PartialRow row, Type type, String columnName, Object value) throws IOException {
try {
switch (type) {
case STRING:
row.addString(columnName, value.toString());
return;
case INT32:
row.addInt(columnName, Integer.valueOf(value.toString()));
return;
case INT64:
row.addLong(columnName, Long.valueOf(value.toString()));
return;
case DOUBLE:
row.addDouble(columnName, Double.valueOf(value.toString()));
return;
case BOOL:
row.addBoolean(columnName, (Boolean) value);
return;
case INT8:
row.addByte(columnName, (byte) value);
return;
case INT16:
row.addShort(columnName, (short) value);
return;
case BINARY:
os.writeObject(value);
row.addBinary(columnName, out.toByteArray());
return;
case FLOAT:
row.addFloat(columnName, Float.valueOf(String.valueOf(value)));
return;
default:
throw new UnsupportedOperationException("Unknown type " + type);
}
} catch (Exception e) {
logger.error("數(shù)據(jù)插入異常", e);
}
}
}
測(cè)試樣例（這里使用了一個(gè) UserInfo 的 pojo 類，包括 userid、name、age 三個(gè)屬性，文內(nèi)省略了）
package test;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.HashMap;
import java.util.Map;
public class SinkTest {
public static void main(String []args) throws Exception {
// 初始化 flink 執(zhí)行環(huán)境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 生成數(shù)據(jù)源
DataStreamSource<UserInfo> dataSource = env.fromElements(new UserInfo("001", "Jack", 18),
new UserInfo("002", "Rose", 20),
new UserInfo("003", "Cris", 22),
new UserInfo("004", "Lily", 19),
new UserInfo("005", "Lucy", 21),
new UserInfo("006", "Json", 24));
// 轉(zhuǎn)換數(shù)據(jù) map
SingleOutputStreamOperator<Map<String, Object>> mapSource = dataSource.map(new MapFunction<UserInfo, Map<String, Object>>() {
@Override
public Map<String, Object> map(UserInfo value) throws Exception {
Map<String, Object> map = new HashMap<>();
map.put("userid", value.userid);
map.put("name", value.name);
map.put("age", value.age);
return map;
}
});
// sink 到 kudu
String kuduMaster = "host";
String tableInfo = "tablename";
mapSource.addSink(new SinkKudu(kuduMaster, tableInfo));
env.execute("sink-test");
}
}
小結(jié)
- 這里自定義 SinkKudu 函數(shù)，通過一個(gè)簡(jiǎn)單樣例進(jìn)行測(cè)試。當(dāng)然，這里的 source 可以換成讀取 kafka 數(shù)據(jù)進(jìn)行流式數(shù)據(jù)的處理。flink 讀取 kafka，然后寫入 kudu，是生產(chǎn)中實(shí)時(shí) ETL 經(jīng)常采用的方案。
