我正在尝试将方解石与Kafka集成,我引用了CsvStreamableTable。
使用fowlloing代码将每个ConsumerRecord转换为Object[]:
static class ArrayRowConverter extends RowConverter<Object[]> {
private List<Schema.Field> fields;
public ArrayRowConverter(List<Schema.Field> fields) {
this.fields = fields;
}
@Override
Object[] convertRow(ConsumerRecord<String, GenericRecord> consumerRecord) {
Object[] objects = new Object[fields.size()+1];
int i = 0 ;
objects[i++] = consumerRecord.timestamp();
for(Schema.Field field : this.fields) {
Object obj = consumerRecord.value().get(field.name());
if( obj instanceof Utf8 ){
objects[i ++] = obj.toString();
}else {
objects[i ++] = obj;
}
}
return objects;
}
}枚举器的实现如下所示,一个线程不断轮询来自kafka的记录,并将它们放入一个队列,getRecord()方法从该队列轮询:
public E current() {
return current;
}
public boolean moveNext() {
for(;;) {
if(cancelFlag.get()) {
return false;
}
ConsumerRecord<String, GenericRecord> record = getRecord();
if(record == null) {
try {
Thread.sleep(200L);
} catch (InterruptedException e) {
e.printStackTrace();
}
continue;
}
current = rowConvert.convertRow(record);
return true;
}
}我测试了SELECT STREAM * FROM Kafka.clicks,它工作得很好。rowtime是显式添加的第一列,取值为Kafka的记录时间戳。
但当我试着
SELECT STREAM FLOOR(rowtime TO HOUR)
AS rowtime,ip,COUNT(*) AS c FROM KAFKA.clicks GROUP BY FLOOR(rowtime TO HOUR), ip它抛出异常
java.sql.SQLException: Error while executing SQL "SELECT STREAM FLOOR(rowtime TO HOUR) AS rowtime,ip,COUNT(*) AS c FROM KAFKA.clicks GROUP BY FLOOR(rowtime TO HOUR), ip": From line 1, column 85 to line 1, column 119: Streaming aggregation requires at least one monotonic expression in GROUP BY clause
at org.apache.calcite.avatica.Helper.createException(Helper.java:56)
at org.apache.calcite.avatica.Helper.createException(Helper.java:41)发布于 2017-02-24 08:56:18
您需要声明"ROWTIME“列是单调的。在MockCatalogReader中,请注意“订单”和“发货”流中的"ROWTIME“是如何声明为单调的。这就是为什么SqlValidatorTest.testStreamGroupBy()中的一些查询是有效的,而另一些查询是无效的。验证器依赖的关键方法是SqlValidatorTable.getMonotonicity(String columnName)。
https://stackoverflow.com/questions/42423559
复制相似问题