今天教大家借助一款框架快速实现一个数据库，分钟这个框架就是教写据库Calcite，下面会带大家通过两个例子快速教会大家怎么实现，个数一个是分钟可以通过 SQL 语句的方式可以直接查询文件内容，第二个是教写据库模拟 Mysql 查询功能，以及最后告诉大家怎么实现 SQL 查询 Kafka 数据。个数
Calcite 是分钟一个用于优化异构数据源的查询处理的可插拔基础框架（他是一个框架），可以将任意数据（Any data,教写据库 Anywhere）DML 转换成基于 SQL 的 DML 引擎，并且我们可以选择性的个数使用它的部分功能。
当我们需要自建一个数据库的时候，数据可以为任何格式的，比如text、word、xml、mysql、es、csv、第三方接口数据等等，我们只有数据，我们想让这些数据支持 SQL 形式动态增删改查。
另外，像Hive、Drill、Flink、Phoenix 和 Storm 等项目中，数据处理系统都是使用 Calcite 来做 SQL 解析和查询优化，当然，还有部分用来构建自己的 JDBC driver。
就是将标准 SQL（可以理解为Mysql）关键词以及关键词之间的字符串截取出来，每一个token，会被封装为一个SqlNode，SqlNode会衍生很多子类，比如Select会被封装为SqlSelect，当前 SqlNode 也能反解析为 SQL 文本。
某个字段的名称和类型信息
多个 RelDataTypeField 组成了 RelDataType，可以理解为数据行
一个完整的表的信息
所有元数据的组合，可以理解为一组 Table 或者库的概念
<dependency>
<groupId>org.apache.calcite</groupId>
<artifactId>calcite-core</artifactId>
<!-- 目前最新版本 2022-09-10日更新-->
<version>1.32.0</version>
</dependency>
model.json 里面主要描述或者说告诉 Calcite 如何创建 Schema，也就是告诉框架怎么创建出库。
{
"version": "1.0",
"defaultSchema": "CSV",
"schemas": [
{
"name": "CSV",
"type": "custom",
"factory": "csv.CsvSchemaFactory",
"operand": {
"directory": "csv"
}
}
]
}接下来还需要定义一个 csv 文件，用来定义表结构。
NAME:string,MONEY:string
aixiaoxian,10000万
xiaobai,10000万
adong,10000万
maomao,10000万
xixi,10000万
zizi,10000万
wuwu,10000万
kuku,10000万
整个项目的结构大概就是这样：
在上述文件中指定的包路径下去编写 CsvSchemaFactory 类，实现 SchemaFactory 接口，并且实现里面唯一的方法 create 方法，创建Schema（库）。
public class CsvSchemaFactory implements SchemaFactory {
@Override
public Schema create(SchemaPlus parentSchema, String name,
Map<String, Object> operand) {
final String directory = (String) operand.get("directory");
File directoryFile = new File(directory);
return new CsvSchema(directoryFile, "scannable");
}
}有了 SchemaFactory，接下来需要自定义 Schema 类。
自定义的 Schema 需要实现 Schema 接口，但是直接实现要实现的方法太多，我们去实现官方的 AbstractSchema 类，这样就只需要实现一个方法就行（如果有其他定制化需求可以实现原生接口）。
核心的逻辑就是createTableMap方法，用于创建出 Table 表。
他会扫描指定的Resource下面的所有 csv 文件，将每个文件映射成Table对象，最终以map形式返回，Schema接口的其他几个方法会用到这个对象。
@Override
protected Map<String, Table> getTableMap() {
if (tableMap == null) {
tableMap = createTableMap();
}
return tableMap;
}
private Map<String, Table> createTableMap() {
final Source baseSource = Sources.of(directoryFile);
File[] files = directoryFile.listFiles((dir, name) -> {
final String nameSansGz = trim(name, ".gz");
return nameSansGz.endsWith(".csv");
});
if (files == null) {
System.out.println("directory " + directoryFile + " not found");
files = new File[0];
}
final ImmutableMap.Builder<String, Table> builder = ImmutableMap.builder();
for (File file : files) {
Source source = Sources.of(file);
final Source sourceSansCsv = source.trimOrNull(".csv");
if (sourceSansCsv != null) {
final Table table = createTable(source);
builder.put(sourceSansCsv.relative(baseSource).path(), table);
}
}
return builder.build();
}
Schema 有了，并且数据文件 csv 也映射成 Table 了，一个 csv 文件对应一个 Table。
接下来我们去自定义 Table，自定义 Table 的核心是我们要定义字段的类型和名称，以及如何读取 csv文件。
先获取数据类型和名称，即单表结构，从csv文件头中获取（当前文件头需要我们自己定义，包括规则我们也可以定制化）。
public abstract class CsvTable extends AbstractTable {
protected final Source source;
protected final @Nullable RelProtoDataType protoRowType;
private @Nullable RelDataType rowType;
private @Nullable List<RelDataType> fieldTypes;
CsvTable(Source source, @Nullable RelProtoDataType protoRowType) {
this.source = source;
this.protoRowType = protoRowType;
}
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory) {
if (protoRowType != null) {
return protoRowType.apply(typeFactory);
}
if (rowType == null) {
rowType = CsvEnumerator.deduceRowType((JavaTypeFactory) typeFactory, source,
null);
}
return rowType;
}
public List<RelDataType> getFieldTypes(RelDataTypeFactory typeFactory) {
if (fieldTypes == null) {
fieldTypes = new ArrayList<>();
CsvEnumerator.deduceRowType((JavaTypeFactory) typeFactory, source,
fieldTypes);
}
return fieldTypes;
}
public static RelDataType deduceRowType(JavaTypeFactory typeFactory,
Source source, @Nullable List<RelDataType> fieldTypes) {
final List<RelDataType> types = new ArrayList<>();
final List<String> names = new ArrayList<>();
try (CSVReader reader = openCsv(source)) {
String[] strings = reader.readNext();
if (strings == null) {
strings = new String[]{ "EmptyFileHasNoColumns:boolean"};
}
for (String string : strings) {
final String name;
final RelDataType fieldType;
final int colon = string.indexOf(':');
if (colon >= 0) {
name = string.substring(0, colon);
String typeString = string.substring(colon + 1);
Matcher decimalMatcher = DECIMAL_TYPE_PATTERN.matcher(typeString);
if (decimalMatcher.matches()) {
int precision = Integer.parseInt(decimalMatcher.group(1));
int scale = Integer.parseInt(decimalMatcher.group(2));
fieldType = parseDecimalSqlType(typeFactory, precision, scale);
} else {
switch (typeString) {
case "string":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.VARCHAR);
break;
case "boolean":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.BOOLEAN);
break;
case "byte":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.TINYINT);
break;
case "char":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.CHAR);
break;
case "short":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.SMALLINT);
break;
case "int":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.INTEGER);
break;
case "long":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.BIGINT);
break;
case "float":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.REAL);
break;
case "double":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.DOUBLE);
break;
case "date":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.DATE);
break;
case "timestamp":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.TIMESTAMP);
break;
case "time":
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.TIME);
break;
default:
LOGGER.warn(
"Found unknown type: { } in file: { } for column: { }. Will assume the type of "
+ "column is string.",
typeString, source.path(), name);
fieldType = toNullableRelDataType(typeFactory, SqlTypeName.VARCHAR);
break;
}
}
} else {
name = string;
fieldType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
}
names.add(name);
types.add(fieldType);
if (fieldTypes != null) {
fieldTypes.add(fieldType);
}
}
} catch (IOException e) {
}
if (names.isEmpty()) {
names.add("line");
types.add(typeFactory.createSqlType(SqlTypeName.VARCHAR));
}
return typeFactory.createStructType(Pair.zip(names, types));
}
}获取文件中的数据，上面把Table的表结构字段名称和类型都获取到了以后，就剩最后一步了，获取文件中的数据。我们需要自定义一个类，实现 ScannableTable 接口，并且实现里面唯一的方法 scan 方法，其实本质上就是读文件，然后把文件的每一行的数据和上述获取的 fileType 进行匹配。
@Override
public Enumerable<Object[]> scan(DataContext root) {
JavaTypeFactory typeFactory = root.getTypeFactory();
final List<RelDataType> fieldTypes = getFieldTypes(typeFactory);
final List<Integer> fields = ImmutableIntList.identity(fieldTypes.size());
final AtomicBoolean cancelFlag = DataContext.Variable.CANCEL_FLAG.get(root);
return new AbstractEnumerable<@Nullable Object[]>() {
@Override
public Enumerator<@Nullable Object[]> enumerator() {
return new CsvEnumerator<>(source, cancelFlag, false, null,
CsvEnumerator.arrayConverter(fieldTypes, fields, false));
}
};
}
public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream,
@Nullable String @Nullable [] filterValues, RowConverter<E> rowConverter) {
this.cancelFlag = cancelFlag;
this.rowConverter = rowConverter;
this.filterValues = filterValues == null ? null
: ImmutableNullableList.copyOf(filterValues);
try {
this.reader = openCsv(source);
this.reader.readNext();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public E current() {
return castNonNull(current);
}
@Override
public boolean moveNext() {
try {
outer:
for (; ; ) {
if (cancelFlag.get()) {
return false;
}
final String[] strings = reader.readNext();
if (strings == null) {
current = null;
reader.close();
return false;
}
if (filterValues != null) {
for (int i = 0; i < strings.length; i++) {
String filterValue = filterValues.get(i);
if (filterValue != null) {
if (!filterValue.equals(strings[i])) {
continue outer;
}
}
}
}
current = rowConverter.convertRow(strings);
return true;
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
protected @Nullable Object convert(@Nullable RelDataType fieldType, @Nullable String string) {
if (fieldType == null || string == null) {
return string;
}
switch (fieldType.getSqlTypeName()) {
case BOOLEAN:
if (string.length() == 0) {
return null;
}
return Boolean.parseBoolean(string);
case TINYINT:
if (string.length() == 0) {
return null;
}
return Byte.parseByte(string);
case SMALLINT:
if (string.length() == 0) {
return null;
}
return Short.parseShort(string);
case INTEGER:
if (string.length() == 0) {
return null;
}
return Integer.parseInt(string);
case BIGINT:
if (string.length() == 0) {
return null;
}
return Long.parseLong(string);
case FLOAT:
if (string.length() == 0) {
return null;
}
return Float.parseFloat(string);
case DOUBLE:
if (string.length() == 0) {
return null;
}
return Double.parseDouble(string);
case DECIMAL:
if (string.length() == 0) {
return null;
}
return parseDecimal(fieldType.getPrecision(), fieldType.getScale(), string);
case DATE:
if (string.length() == 0) {
return null;
}
try {
Date date = TIME_FORMAT_DATE.parse(string);
return (int) (date.getTime() / DateTimeUtils.MILLIS_PER_DAY);
} catch (ParseException e) {
return null;
}
case TIME:
if (string.length() == 0) {
return null;
}
try {
Date date = TIME_FORMAT_TIME.parse(string);
return (int) date.getTime();
} catch (ParseException e) {
return null;
}
case TIMESTAMP:
if (string.length() == 0) {
return null;
}
try {
Date date = TIME_FORMAT_TIMESTAMP.parse(string);
return date.getTime();
} catch (ParseException e) {
return null;
}
case VARCHAR:
default:
return string;
}
}
6. 最后
至此我们需要准备的东西：库、表名称、字段名称、字段类型都有了，接下来我们去写我们的 SQL 语句查询我们的数据文件。
创建好几个测试的数据文件，例如上面项目结构中我创建 2 个 csv 文件USERINFO.csv、ASSET.csv，然后创建测试类。
这样跑起来，就可以通过 SQL 语句的方式直接查询数据了。
public class Test {
public static void main(String[] args) throws SQLException {
Connection connection = null;
Statement statement = null;
try {
Properties info = new Properties();
info.put("model", Sources.of(Test.class.getResource("/model.json")).file().getAbsolutePath());
connection = DriverManager.getConnection("jdbc:calcite:", info);
statement = connection.createStatement();
print(statement.executeQuery("select * from asset "));
print(statement.executeQuery(" select * from userinfo "));
print(statement.executeQuery(" select age from userinfo where name ='aixiaoxian' "));
print(statement.executeQuery(" select * from userinfo where age >60 "));
print(statement.executeQuery(" select * from userinfo where name like 'a%' "));
} finally {
connection.close();
}
}
private static void print(ResultSet resultSet) throws SQLException {
final ResultSetMetaData metaData = resultSet.getMetaData();
final int columnCount = metaData.getColumnCount();
while (resultSet.next()) {
for (int i = 1; ; i++) {
System.out.print(resultSet.getString(i));
if (i < columnCount) {
System.out.print(", ");
} else {
System.out.println();
break;
}
}
}
}
}查询结果：
这里在测试的时候踩到2个坑，大家如果自己实验的时候可以避免下。
Calcite 默认会把你的 SQL 语句中的表名和类名全部转换为大写，因为默认的 csv（其他文件也一样）文件的名称就是表名，除非你自定义规则，所以你的文件名要写成大写。
Calcite 有一些默认的关键字不能用作表名，不然会查询失败，比如我刚开始定的user.csv就一直查不出来，改成USERINFO就可以了，这点和Mysql 的内置关键字差不多，也可以通过个性化配置去改。
首先，还是先准备Calcite 需要的东西：库、表名称、字段名称、字段类型。
如果数据源使用Mysql的话，这些都不用我们去 JAVA 服务中去定义，直接在 Mysql 客户端创建好，这里直接创建两张表用于测试，就和我们的csv文件一样。
CREATE TABLE `USERINFO1` (
`NAME` varchar(255) CHARACTER SET utf8mb3 COLLATE utf8_general_ci DEFAULT NULL,
`AGE` int DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;
CREATE TABLE `ASSET` (
`NAME` varchar(255) CHARACTER SET utf8mb3 COLLATE utf8_general_ci DEFAULT NULL,
`MONEY` varchar(255) CHARACTER SET utf8mb3 COLLATE utf8_general_ci DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;
上述csv 案例中的 SchemaFactory 以及 Schema 这些都不需要创建，因为 Calcite 默认提供了 Mysql 的 Adapter适配器。
其实，上述两步都不需要做，我们真正要做的是，告诉Calcite 你的 JDBC 的连接信息就行了，也是在 model.json 文件中定义。
{
"version": "1.0",
"defaultSchema": "Demo",
"schemas": [
{
"name": "Demo",
"type": "custom",
"factory": "org.apache.calcite.adapter.jdbc.JdbcSchema$Factory",
"operand": {
"jdbcDriver": "com.mysql.cj.jdbc.Driver",
"jdbcUrl": "jdbc:mysql://localhost:3306/irving",
"jdbcUser": "root",
"jdbcPassword": "123456"
}
}
]
}在项目中引入 Mysql 的驱动包
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.30</version>
</dependency>
写好测试类，这样直接就相当于完成了所有的功能了。
public class TestMysql {
public static void main(String[] args) throws SQLException {
Connection connection = null;
Statement statement = null;
try {
Properties info = new Properties();
info.put("model", Sources.of(TestMysql.class.getResource("/mysqlmodel.json")).file().getAbsolutePath());
connection = DriverManager.getConnection("jdbc:calcite:", info);
statement = connection.createStatement();
statement.executeUpdate(" insert into userinfo1 values ('xxx',12) ");
print(statement.executeQuery("select * from asset "));
print(statement.executeQuery(" select * from userinfo1 "));
print(statement.executeQuery(" select age from userinfo1 where name ='aixiaoxian' "));
print(statement.executeQuery(" select * from userinfo1 where age >60 "));
print(statement.executeQuery(" select * from userinfo1 where name like 'a%' "));
} finally {
connection.close();
}
}
private static void print(ResultSet resultSet) throws SQLException {
final ResultSetMetaData metaData = resultSet.getMetaData();
final int columnCount = metaData.getColumnCount();
while (resultSet.next()) {
for (int i = 1; ; i++) {
System.out.print(resultSet.getString(i));
if (i < columnCount) {
System.out.print(", ");
} else {
System.out.println();
break;
}
}
}
}
}查询结果：
上述我们在 model.json 文件中指定了org.apache.calcite.adapter.jdbc.JdbcSchema$Factory类，可以看下这个类的代码。
这个类是把 Factory 和 Schema 写在了一起，其实就是调用schemafactory类的create方法创建一个 schema 出来，和我们上面自定义的流程是一样的。
其中JdbcSchema类也是 Schema 的子类，所以也会实现getTable方法（这个我们上述也实现了，我们当时是获取表结构和表的字段类型以及名称，是从csv文件头中读文件的），JdbcSchema的实现是通过连接 Mysql 服务端查询元数据信息，再将这些信息封装成 Calcite需要的对象格式。
这里同样要注意 csv方式的2个注意点，大小写和关键字问题。
public static JdbcSchema create(
SchemaPlus parentSchema,
String name,
Map<String, Object> operand) {
DataSource dataSource;
try {
final String dataSourceName = (String) operand.get("dataSource");
if (dataSourceName != null) {
dataSource =
AvaticaUtils.instantiatePlugin(DataSource.class, dataSourceName);
} else {
final String jdbcUrl = (String) requireNonNull(operand.get("jdbcUrl"), "jdbcUrl");
final String jdbcDriver = (String) operand.get("jdbcDriver");
final String jdbcUser = (String) operand.get("jdbcUser");
final String jdbcPassword = (String) operand.get("jdbcPassword");
dataSource = dataSource(jdbcUrl, jdbcDriver, jdbcUser, jdbcPassword);
}
} catch (Exception e) {
throw new RuntimeException("Error while reading dataSource", e);
}
String jdbcCatalog = (String) operand.get("jdbcCatalog");
String jdbcSchema = (String) operand.get("jdbcSchema");
String sqlDialectFactory = (String) operand.get("sqlDialectFactory");
if (sqlDialectFactory == null || sqlDialectFactory.isEmpty()) {
return JdbcSchema.create(
parentSchema, name, dataSource, jdbcCatalog, jdbcSchema);
} else {
SqlDialectFactory factory = AvaticaUtils.instantiatePlugin(
SqlDialectFactory.class, sqlDialectFactory);
return JdbcSchema.create(
parentSchema, name, dataSource, factory, jdbcCatalog, jdbcSchema);
}
}
@Override public @Nullable Table getTable(String name) {
return getTableMap(false).get(name);
}
private synchronized ImmutableMap<String, JdbcTable> getTableMap(
boolean force) {
if (force || tableMap == null) {
tableMap = computeTables();
}
return tableMap;
}
private ImmutableMap<String, JdbcTable> computeTables() {
Connection connection = null;
ResultSet resultSet = null;
try {
connection = dataSource.getConnection();
final Pair<@Nullable String, @Nullable String> catalogSchema = getCatalogSchema(connection);
final String catalog = catalogSchema.left;
final String schema = catalogSchema.right;
final Iterable<MetaImpl.MetaTable> tableDefs;
Foo threadMetadata = THREAD_METADATA.get();
if (threadMetadata != null) {
tableDefs = threadMetadata.apply(catalog, schema);
} else {
final List<MetaImpl.MetaTable> tableDefList = new ArrayList<>();
final DatabaseMetaData metaData = connection.getMetaData();
resultSet = metaData.getTables(catalog, schema, null, null);
while (resultSet.next()) {
final String catalogName = resultSet.getString(1);
final String schemaName = resultSet.getString(2);
final String tableName = resultSet.getString(3);
final String tableTypeName = resultSet.getString(4);
tableDefList.add(
new MetaImpl.MetaTable(catalogName, schemaName, tableName,
tableTypeName));
}
tableDefs = tableDefList;
}
final ImmutableMap.Builder<String, JdbcTable> builder =
ImmutableMap.builder();
for (MetaImpl.MetaTable tableDef : tableDefs) {
final String tableTypeName2 =
tableDef.tableType == null
? null
: tableDef.tableType.toUpperCase(Locale.ROOT).replace(' ', '_');
final TableType tableType =
Util.enumVal(TableType.OTHER, tableTypeName2);
if (tableType == TableType.OTHER && tableTypeName2 != null) {
System.out.println("Unknown table type: " + tableTypeName2);
}
final JdbcTable table =
new JdbcTable(this, tableDef.tableCat, tableDef.tableSchem,
tableDef.tableName, tableType);
builder.put(tableDef.tableName, table);
}
return builder.build();
} catch (SQLException e) {
throw new RuntimeException(
"Exception while reading tables", e);
} finally {
close(connection, null, resultSet);
}
}
OK，到这里基本上两个简单的案例已经演示好了，最后补充一下整个Calcite架构和整个 SQL 的执行流程。
整个流程如下：SQL解析（Parser）=> SQL校验（Validator）=> SQL查询优化（optimizer）=> SQL生成 => SQL执行
所有的 SQL 语句在执行前都需要经历 SQL 解析器解析，解析器的工作内容就是将 SQL 中的 Token 解析成抽象语法树，每个树的节点都是一个 SqlNode，这个过程其实就是 Sql Text => SqlNode 的过程。
我们前面的 Demo 没有自定义 Parser，是因为 Calcite 采用了自己默认的 Parser（SqlParserImpl）。
SqlNode是整个解析中的核心，比如图中你可以发现，对于每个比如select、from、where关键字之后的内容其实都是一个SqlNode。
parserConfig方法主要是设置 SqlParserFactory 的参数，比如我们上面所说的我本地测试的时候踩的大小写的坑，就可以在这里设置。
直接调用setCaseSensitive=false即不会将 SQL 语句中的表名列名转为大写，下面是默认的，其他的参数可以按需配置。
SQL 语句先经过 Parser，然后经过语法验证器，注意 Parser 并不会验证语法的正确性。
其实 Parser 只会验证 SQL 关键词的位置是否正确，我们上述2个 Parser 的例子中都没有创建 schema 和 table 这些，但是如果这样写，那就会报错，这个错误就是 parser 检测后抛出来的（ParseLocationErrorTest)。
真正的校验在 validator 中，会去验证查询的表名是否存在，查询的字段是否存在，类型是否匹配，这个过程比较复杂，默认的 validator 是SqlValidatorImpl。
比如关系代数，比如什么投影、笛卡尔积这些，Calcite提供了很多内部的优化器，也可以实现自己的优化器。
Calcite 是不包含存储层的，所以提供一种适配器的机制来访问外部的数据存储或者存储引擎。
官网里面写了未来会支持Kafka适配器到公共Api中，到时候使用起来就和上述集成Mysql一样方便，但是现在还没有支持，我这里给大家提供个自己实现的方式，这样就可以通过 SQL 的方式直接查询 Kafka 中的 Topic 数据等信息。
这里我们内部集成实现了KSQL的能力，查询结果是OK的。
还是像上述步骤一样，我们需要准备库、表名称、字段名称、字段类型、数据源（多出来的地方）。
自定义Sql解析，之前我们都没有自定义解析，这里需要自定义解析，是因为我需要动态解析sql的where条件里面的partation。
适配器获取数据源
public class KafkaConsumerAdapter {
public static List<KafkaResult> executor(KafkaSqlInfo kafkaSql) {
Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaSql.getSeeds());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
List<TopicPartition> topics = new ArrayList<>();
for (Integer partition : kafkaSql.getPartition()) {
TopicPartition tp = new TopicPartition(kafkaSql.getTableName(), partition);
topics.add(tp);
}
consumer.assign(topics);
for (TopicPartition tp : topics) {
Map<TopicPartition, Long> offsets = consumer.endOffsets(Collections.singleton(tp));
long position = 500;
if (offsets.get(tp).longValue() > position) {
consumer.seek(tp, offsets.get(tp).longValue() - 500);
} else {
consumer.seek(tp, 0);
}
}
List<KafkaResult> results = new ArrayList<>();
boolean flag = true;
while (flag) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
KafkaResult result = new KafkaResult();
result.setPartition(record.partition());
result.setOffset(record.offset());
result.setMsg(record.value());
result.setKey(record.key());
results.add(result);
}
if (!records.isEmpty()) {
flag = false;
}
}
consumer.close();
return results;
}
}执行查询，就可以得到我们想要的效果了。
public class TestKafka {
public static void main(String[] args) throws Exception {
KafkaService kafkaService = new KafkaService();
KafkaSqlInfo sqlInfo = kafkaService.parseSql("select * from `cmdb-calltopo` where `partition` in (0,1,2) limit 1000 ");
List<KafkaResult> results = KafkaConsumerAdapter.executor(sqlInfo);
query(sqlInfo.getTableName(), results, sqlInfo.getSql());
sqlInfo = kafkaService.parseSql("select * from `cmdb-calltopo` where `partition` in (0,1,2) AND msg like '%account%' limit 1000 ");
results = KafkaConsumerAdapter.executor(sqlInfo);
query(sqlInfo.getTableName(), results, sqlInfo.getSql());
sqlInfo = kafkaService.parseSql("select count(*) AS addad from `cmdb-calltopo` where `partition` in (0,1,2) limit 1000 ");
results = KafkaConsumerAdapter.executor(sqlInfo);
query(sqlInfo.getTableName(), results, sqlInfo.getSql());
}
private static void query(String tableName, List<KafkaResult> results,
String sql) throws Exception {
String model = createTempJson();
KafkaTableSchema.generateSchema(tableName, results);
Properties info = new Properties();
info.setProperty("lex", Lex.JAVA.toString());
Connection connection = DriverManager.getConnection(Driver.CONNECT_STRING_PREFIX + "model=inline:" + model, info);
Statement st = connection.createStatement();
ResultSet result = st.executeQuery(sql);
ResultSetMetaData rsmd = result.getMetaData();
List<Map<String, Object>> ret = new ArrayList<>();
while (result.next()) {
Map<String, Object> map = new LinkedHashMap<>();
for (int i = 1; i <= rsmd.getColumnCount(); i++) {
map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i)));
}
ret.add(map);
}
result.close();
st.close();
connection.close();
}
private static void print(ResultSet resultSet) throws SQLException {
final ResultSetMetaData metaData = resultSet.getMetaData();
final int columnCount = metaData.getColumnCount();
while (resultSet.next()) {
for (int i = 1; ; i++) {
System.out.print(resultSet.getString(i));
if (i < columnCount) {
System.out.print(", ");
} else {
System.out.println();
break;
}
}
}
}
private static String createTempJson() throws IOException {
JSONObject object = new JSONObject();
object.put("version", "1.0");
object.put("defaultSchema", "QAKAFKA");
JSONArray array = new JSONArray();
JSONObject tmp = new JSONObject();
tmp.put("name", "QAKAFKA");
tmp.put("type", "custom");
tmp.put("factory", "kafka.KafkaSchemaFactory");
array.add(tmp);
object.put("schemas", array);
return object.toJSONString();
}
}生成临时的model.json，之前是基于文件，现在基于text字符串，mode=inline模式
设置我的表结构、表名称、字段名、字段类型等，并放置在内存中，同时将适配器查询出来的数据也放进去table里面
获取连接，执行查询，完美！
(责任编辑：热点)
中国金融投资管理(00605.HK)公布消息：将考虑向罗锐及关雪玲提起法律诉讼
中国金融投资管理(00605.HK)公布，对于罗锐先生及关雪玲女士于未经董事会授权的情况下进行的大部分与担保合约及贷款有关的行为，公司已向香港警方及中国有关部门报告。公司将考虑向罗先生及关女士提起法律
...[详细]
华为P40系列发布上市之后，下一部旗舰机就轮要到Mate50系列了，虽然现在华为的处境如何大家都心知肚明，但华为方面也曾多次公开表示Mate50是一定存在的。近日，海外一名渲染师就根据目前网络上流出的
...[详细]
国庆七天长假来临,和家人、朋友一起去哪度假?广州融创文旅城在国庆期间推出“我的潮代”国潮主题系列活动,寰球美食潮饮嘉年华、国风大赏表演、中华神兽巡游等丰富多彩的潮玩活动重磅来袭
...[详细]
稻垣理一郎原作、Boichi作画的同名人气漫画《Dr.STONE石纪元）》TV动画官方公开了第三季全新预告，并表示将于4月6日正式开播。《石纪元：新世界》第3季新预告：《Dr. STONE》是稻垣理一
...[详细]碧桂园服务(06098.HK)公布：拟收购蓝光嘉宝服务(02606.HK)64.62%股权 明日复牌
碧桂园服务(06098.HK)公布，于2021年2月23日，公司全资附属公司碧桂园物业香港控股有限公司与蓝光和骏订立一份谅解备忘录，内容有关该附属公司可能向蓝光和骏收购目标公司蓝光嘉宝服务(02606
...[详细]
11月24日，富途控股(Nasdaq： FUTU，下称“富途”)公布了2021年第三季度未经审计财报。财报显示，富途三季度总营收达2.2亿美元(约17.3亿港元)，同比增长83.0%，非美国通用会计准
...[详细]博元电声ANGEL 3：分享好品质个性音响，分享最真实的声音
纵观全球音频产业，像博元电声这般以摧枯拉朽之势占据音响领域中高端地位的新生品牌屈指可数 。博元电声是一家深耕专业级音响的民族品牌，一直致力于研发高品质的音响，在声学方面的造诣让同行望尘莫及。运营至今，
...[详细]
近日，所托瑞安与太平洋财产保险宣布达成战略合作，以科技赋能保险，打造商用车风控新模式，共同履行社会责任，为守护道路交通安全共享力量。太平洋财产保险从2021年7月份开始，在承保高风险营运货车业务时，将
...[详细]
在发展事业的过程中，对于一些家底不是很雄厚的个体户来说，想要扩大经营规模，免不了要去申请贷款。支付宝旗下的网商贷，就是一种经营性贷款。网商贷额度突然降低了怎么办?这样做或许能快速恢复额度!网商贷额度突
...[详细]
之前B社宣布《星空》延期至9月6日发售，登陆Xbox和PC平台。官方还发布新视频，展示了游戏场景和画面。虽然游戏片段很短，但也让粉丝们很满意。尽管新视频中游戏场景分辨率不高，但有许多人注意到《星空》诸
...[详细]
10月份全国服务业生产指数同比增长3.8% 总体保持恢复态势
昆汀最后一部电影片名曝光 今年秋季开拍
1499元起！realme Q3s发布：骁龙778G+144Hz高刷屏
波司登再度携手杨幂推出2021时尚系列 引领羽绒服时尚趋势
节能元件(08231.HK)发布公告:预计年度由亏转盈60万美元
