https://flink.apache.org/ logo
Join Slack
Powered by
I made this function for calculating the median wi...
# troubleshooting
a
I made this function for calculating the median with the t-digest library written in Java. It doesn't work 😭.
Copy code
package fi.elisa.datalake.flink.flinktools.aggregation;

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.annotation.DataTypeHint;
import static org.apache.flink.table.api.Expressions.*;

import com.tdunning.math.stats.MergingDigest;

public class MergingDigestMedian extends AggregateFunction<Double, MergingDigest> {

  @Override
  public @DataTypeHint("RAW") MergingDigest createAccumulator() {
    return new MergingDigest(100); // 100 is a common value for normal uses.
  }

  @Override
  public Double getValue(@DataTypeHint("RAW") MergingDigest acc) {
    return acc.quantile(0.5);
  }

  public void accumulate(@DataTypeHint("RAW") MergingDigest acc, Double value) {
    acc.add(value);
  }

}
I'm calling it in pyflink code by defining it as a temporary system function.
Copy code
table_env.create_java_temporary_system_function("udf_median", "fi.elisa.datalake.flink.flinktools.aggregation.MergingDigestMedian")
It fails. I'll put the logs in the comments.
Copy code
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/flink_1.16.0/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 836, in execute_sql
  File "/opt/flink_1.16.0/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py", line 1321, in __call__
  File "/opt/flink_1.16.0/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 146, in deco
  File "/opt/flink_1.16.0/opt/python/py4j-0.10.9.3-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o60.executeSql.
: org.apache.flink.table.api.ValidationException: SQL validation failed. An error occurred in the type inference logic of function 'udf_median'.
        at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:186)
        at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:113)
        at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertViewQuery(SqlToOperationConverter.java:1118)
        at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertCreateView(SqlToOperationConverter.java:1095)
        at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertValidatedSqlNode(SqlToOperationConverter.java:319)
        at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:262)
        at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:723)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
        at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
        at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.flink.table.api.ValidationException: An error occurred in the type inference logic of function 'udf_median'.
        at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToBridgingSqlFunction(FunctionCatalogOperatorTable.java:128)
        at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToSqlFunction(FunctionCatalogOperatorTable.java:112)
        at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lambda$lookupOperatorOverloads$0(FunctionCatalogOperatorTable.java:92)
        at java.base/java.util.Optional.flatMap(Optional.java:294)
        at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:92)
        at org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1183)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1200)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:945)
        at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704)
        at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:182)
        ... 18 more
Caused by: org.apache.flink.table.api.ValidationException: Could not extract a valid type inference for function class 'fi.elisa.datalake.flink.flinktools.aggregation.MergingDigestMedian'. Please check for implementation mistakes and/or provide a corresponding hint.
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInference(TypeInferenceExtractor.java:150)
        at org.apache.flink.table.types.extraction.TypeInferenceExtractor.forAggregateFunction(TypeInferenceExtractor.java:98)
        at org.apache.flink.table.functions.AggregateFunction.getTypeInference(AggregateFunction.java:213)
        at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToBridgingSqlFunction(FunctionCatalogOperatorTable.java:125)
        ... 30 more
Caused by: org.apache.flink.table.api.ValidationException: Error in extracting a signature to accumulator mapping.
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractAccumulatorMapping(FunctionMappingExtractor.java:135)
        at org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInferenceOrError(TypeInferenceExtractor.java:168)
        at org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInference(TypeInferenceExtractor.java:148)
        ... 33 more
Caused by: org.apache.flink.table.api.ValidationException: Unable to extract a type inference from method:
public void fi.elisa.datalake.flink.flinktools.aggregation.MergingDigestMedian.accumulate(com.tdunning.math.stats.MergingDigest,java.lang.Double)
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractResultMappings(FunctionMappingExtractor.java:183)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractAccumulatorMapping(FunctionMappingExtractor.java:124)
        ... 35 more
Caused by: org.apache.flink.table.api.ValidationException: Could not extract a data type from 'class com.tdunning.math.stats.MergingDigest' in generic class 'org.apache.flink.table.functions.AggregateFunction' in class fi.elisa.datalake.flink.flinktools.aggregation.MergingDigestMedian. Please pass the required data type manually or allow RAW types.
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRawWithTemplate(DataTypeExtractor.java:240)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRaw(DataTypeExtractor.java:218)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeWithClassContext(DataTypeExtractor.java:194)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractFromGeneric(DataTypeExtractor.java:124)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.lambda$createGenericResultExtraction$13(FunctionMappingExtractor.java:478)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.putExtractedResultMappings(FunctionMappingExtractor.java:319)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.collectMethodMappings(FunctionMappingExtractor.java:269)
        at org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractResultMappings(FunctionMappingExtractor.java:169)
        ... 36 more
Caused by: org.apache.flink.table.api.ValidationException: Could not extract a data type from 'class com.tdunning.math.stats.MergingDigest'. Interpreting it as a structured type was also not successful.
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrError(DataTypeExtractor.java:290)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRawWithTemplate(DataTypeExtractor.java:232)
        ... 43 more
Caused by: org.apache.flink.table.api.ValidationException: Field 'mergeCount' of class 'com.tdunning.math.stats.MergingDigest' is neither publicly accessible nor does it have a corresponding getter method.
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:362)
        at org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:357)
        at org.apache.flink.table.types.extraction.ExtractionUtils.validateStructuredFieldReadability(ExtractionUtils.java:522)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.lambda$extractStructuredType$0(DataTypeExtractor.java:492)
        at java.base/java.util.stream.MatchOps$1MatchSink.accept(MatchOps.java:90)
        at java.base/java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1632)
        at java.base/java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:127)
        at java.base/java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:502)
        at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:488)
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
        at java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230)
        at java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196)
        at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
        at java.base/java.util.stream.ReferencePipeline.allMatch(ReferencePipeline.java:533)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractStructuredType(DataTypeExtractor.java:490)
        at org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrError(DataTypeExtractor.java:288)
        ... 44 more
6 Views
Open in Slack
PreviousNext
Powered by