Блог

Попытка использовать агрегатную функцию ApproximateDistinct.ApproximateDistinctFn из apache beam sql, это не удалось.

мой SQL:

 SELECT
    ApproximateDistinct(user_id) as distinct_count,
    profile,
    country_code,
    FROM PCOLLECTION
    GROUP BY profile,country_code
  

И мой код в Apache-Beam, который вызывает этот запрос:

   events.apply(SqlTransform.query(query)
                                    .registerUdaf("ApproximateDistinct", ApproximateDistinct.ApproximateDistinctFn.create(StringUtf8Coder.of()).withSparseRepresentation(16)))
  

Это вызывает это исключение:

 aused by: java.lang.ClassCastException: class sun.reflect.generics.reflectiveObjects.TypeVariableImpl cannot be cast to class java.lang.Class (sun.reflect.generics.reflectiveObjects.TypeVariableImpl and java.lang.Class are in module java.base of loader 'bootstrap')
        at org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils.sqlTypeWithAutoCast(CalciteUtils.java:210)
        at org.apache.beam.sdk.extensions.sql.impl.UdafImpl$1.getType(UdafImpl.java:69)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.prepare.CalciteCatalogReader.toOp(CalciteCatalogReader.java:313)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.prepare.CalciteCatalogReader.toOp(CalciteCatalogReader.java:300)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.prepare.CalciteCatalogReader.lambda$lookupOperatorOverloads$3(CalciteCatalogReader.java:269)
        at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
        at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:177)
        at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1654)
        at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
        at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
        at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
        at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
        at java.base/java.util.stream.ReferencePipeline.forEachOrdered(ReferencePipeline.java:502)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.prepare.CalciteCatalogReader.lookupOperatorOverloads(CalciteCatalogReader.java:270)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:72)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1162)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1147)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1176)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1147)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:921)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:632)
        at org.apache.beam.repackaged.beam_sdks_java_extensions_sql.org.apache.calcite.prepare.PlannerImpl.validate(PlannerImpl.java:188)
  

Нашел, как это преодолеть. Возникла проблема с общим типом для кодировщика ввода. Кроме того, вывод функции объединения не является допустимым типом SQL. Итак, я дублирую класс, удаляю обобщения и изменяю тип вывода. И это то, что я получаю:

 public class ApproximateDistinctCardinalityFn
        extends Combine.CombineFn<String, HyperLogLogPlus, Long> {

    private final int p;

    private final int sp;

    private final Coder<String> inputCoder;

\ here copy rest of the code 

}
  

и функции извлечения:

 
    @Override
    public Long extractOutput(HyperLogLogPlus accumulator) {
        return accumulator.cardinality();
    }