大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu - ToB企服应用市场:ToB评测及商务社交产业平台

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>flink-test</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<flink.version>1.11.1</flink.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-client</artifactId>
<version>1.17.0</version>
</dependency>
</dependencies>
</project>

复制代码

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),

复制代码

package icu.wzk.kudu;
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.ObjectOutputStream;
import java.util.Map;
public class MyFlinkSinkToKudu extends RichSinkFunction<Map<String, Object>> {
private final static Logger logger = Logger.getLogger("MyFlinkSinkToKudu");
private KuduClient kuduClient;
private KuduTable kuduTable;
private String kuduMasterAddr;
private String tableName;
private Schema schema;
private KuduSession kuduSession;
private ByteArrayOutputStream out;
private ObjectOutputStream os;
public MyFlinkSinkToKudu(String kuduMasterAddr, String tableName) {
this.kuduMasterAddr = kuduMasterAddr;
this.tableName = tableName;
}
@Override
public void open(Configuration parameters) throws Exception {
out = new ByteArrayOutputStream();
os = new ObjectOutputStream(out);
kuduClient = new KuduClient.KuduClientBuilder(kuduMasterAddr).build();
kuduTable = kuduClient.openTable(tableName);
schema = kuduTable.getSchema();
kuduSession = kuduClient.newSession();
kuduSession.setFlushMode(KuduSession.FlushMode.AUTO_FLUSH_BACKGROUND);
}
@Override
public void invoke(Map<String, Object> map, Context context) throws Exception {
if (null == map) {
return;
}
try {
int columnCount = schema.getColumnCount();
Insert insert = kuduTable.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 (null != response) {
logger.error(response.getRowError().toString());
}
}
} catch (Exception e) {
logger.error(e);
}
}
@Override
public void close() throws Exception {
try {
kuduSession.close();
kuduClient.close();
os.close();
out.close();
} catch (Exception e) {
logger.error(e);
}
}
private void insertData(PartialRow row, Type type, String columnName, Object value) {
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.valueOf(value.toString()));
return;
case BINARY:
os.writeObject(value);
row.addBinary(columnName, out.toByteArray());
return;
case FLOAT:
row.addFloat(columnName, Float.valueOf(value.toString()));
default:
throw new UnsupportedOperationException("Unknown Type: " + type);
}
} catch (Exception e) {
logger.error("插入数据异常: " + e);
}
}
}

复制代码

package icu.wzk.kudu;
public class UserInfo {
private String id;
private String name;
private Integer age;
public UserInfo(String id, String name, Integer age) {
this.id = id;
this.name = name;
this.age = age;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
}

复制代码

package icu.wzk.kudu;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.CreateTableOptions;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import java.util.ArrayList;
import java.util.List;
public class KuduCreateTable {
public static void main(String[] args) throws KuduException {
String masterAddress = "localhost:7051,localhost:7151,localhost:7251";
KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(masterAddress);
KuduClient kuduClient = kuduClientBuilder.build();
String tableName = "user";
List<ColumnSchema> columnSchemas = new ArrayList<>();
ColumnSchema id = new ColumnSchema
.ColumnSchemaBuilder("id", Type.INT32)
.key(true)
.build();
columnSchemas.add(id);
ColumnSchema name = new ColumnSchema
.ColumnSchemaBuilder("name", Type.STRING)
.key(false)
.build();
columnSchemas.add(name);
ColumnSchema age = new ColumnSchema
.ColumnSchemaBuilder("age", Type.INT32)
.key(false)
.build();
columnSchemas.add(age);
Schema schema = new Schema(columnSchemas);
CreateTableOptions options = new CreateTableOptions();
// 副本数量为1
options.setNumReplicas(1);
List<String> colrule = new ArrayList<>();
colrule.add("id");
options.addHashPartitions(colrule, 3);
kuduClient.createTable(tableName, schema, options);
kuduClient.close();
}
}

复制代码

package icu.wzk.kudu;
import org.apache.flink.api.common.functions.MapFunction;
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;
import java.util.stream.Stream;
public class SinkToKuduTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
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)
);
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("id", value.getId());
map.put("name", value.getName());
map.put("age", value.getAge());
return map;
}
});
String kuduMasterAddr = "localhost:7051,localhost:7151,localhost:7251";
String tableInfo = "user";
mapSource.addSink(new MyFlinkSinkToKudu(kuduMasterAddr, tableInfo));
env.execute("SinkToKuduTest");
}
}

复制代码