Java通过calcite实时读取kafka中的数据
引入maven依赖
<dependency>
<groupId>org.apache.calcite</groupId>
<artifactId>calcite-kafka</artifactId>
<version>1.28.0</version>
</dependency>
测试代码
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Properties;
public class CalciteDemo {
public static void main(String[] args) throws SQLException {
String model = "inline:" +
"{\n" +
" \"version\": \"1.0\",\n" +
" \"defaultSchema\": \"KAFKA\",\n" +
" \"schemas\": [\n" +
" {\n" +
" \"name\": \"KAFKA\",\n" +
" \"tables\": [\n" +
" {\n" +
" \"name\": \"TEST_TABLE\",\n" +
" \"factory\": \"org.apache.calcite.adapter.kafka.KafkaTableFactory\",\n" +
" \"stream\": { \"stream\": true },\n" +
" \"operand\": {\n" +
" \"bootstrap.servers\": \"192.168.x.xx:9092\",\n" +
" \"topic.name\": \"my-cloud-events\",\n" +
" \"consumer.params\": {\n" +
" \"group.id\": \"calcite-ut-consumer\",\n" +
" \"key.deserializer\": \"org.apache.kafka.common.serialization.ByteArrayDeserializer\",\n" +
" \"value.deserializer\": \"org.apache.kafka.common.serialization.ByteArrayDeserializer\"\n" +
" }\n" +
" }\n" +
" }\n" +
" ]\n" +
" }\n" +
" ]\n" +
"}";
Properties info = new Properties();
info.put("model", model);
Connection connection = DriverManager.getConnection("jdbc:calcite:", info);
final CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
final String sql7 = "SELECT STREAM * FROM \"KAFKA\".\"TEST_TABLE\"";
print(calciteConnection,sql7);
connection.close();
calciteConnection.close();
}
public static void print(CalciteConnection calciteConnection, String sql7) throws SQLException {
final PreparedStatement statement = calciteConnection.prepareStatement(sql7);
final ResultSet resultSet = statement.executeQuery();
ResultSetMetaData metadata = resultSet.getMetaData();
while (resultSet.next()) {
for (int i = 1; i <= metadata.getColumnCount(); i++) {
System.out.print(metadata.getColumnLabel(i) + "=" + resultSet.getString(i) + ",");
}
System.out.println();
}
}
}
发送测试数据
运行结果
