https://flink.apache.org/ logo
Join Slack
Powered by
Hello , From the query schema as well as Sink sche...
# troubleshooting
n
Hello , From the query schema as well as Sink schema if the attribute 
record_id
is removed then flink jobs loads. In case 
record_id
is added even though the data type for query as well as sink matches it job load fails with the below error. record_id is created by UUID() method as part of flink SQL.
Copy code
Query schema: [resource: STRING NOT NULL, attribute: STRING NOT NULL, duration: STRING, duration_type: STRING NOT NULL, id: STRING, value: STRING NOT NULL, year: STRING, month: STRING, day: STRING, timestamp: TIMESTAMP_LTZ(3) NOT NULL, record_id: CHAR(36) NOT NULL, _c11: TIMESTAMP_LTZ(3) NOT NULL]
Sink schema:  [resource: STRING, id: STRING, duration: STRING, duration_type: STRING, attribute: STRING, value: STRING, year: STRING, month: STRING, day: STRING, record_id: CHAR(36), timestamp: TIMESTAMP_LTZ(3), _c11: TIMESTAMP_LTZ(3)]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:846)
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240)
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1090)
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1168)
	at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1168)
Caused by: org.apache.flink.table.api.ValidationException: Column types of query result and sink for '*anonymous_datastream_sink$1*' do not match.
Cause: Incompatible types for sink column 'record_id' at position 9.

Query schema: [resource: STRING NOT NULL, attribute: STRING NOT NULL, duration: STRING, duration_type: STRING NOT NULL, id: STRING, value: STRING NOT NULL, year: STRING, month: STRING, day: STRING, timestamp: TIMESTAMP_LTZ(3) NOT NULL, record_id: CHAR(36) NOT NULL, _c11: TIMESTAMP_LTZ(3) NOT NULL]
Sink schema:  [resource: STRING, id: STRING, duration: STRING, duration_type: STRING, attribute: STRING, value: STRING, year: STRING, month: STRING, day: STRING, record_id: CHAR(36), timestamp: TIMESTAMP_LTZ(3), _c11: TIMESTAMP_LTZ(3)]
	at org.apache.flink.table.planner.connectors.DynamicSinkUtils.createSchemaMismatchException(DynamicSinkUtils.java:453)
	at org.apache.flink.table.planner.connectors.DynamicSinkUtils.validateSchemaAndApplyImplicitCast(DynamicSinkUtils.java:265)
	at org.apache.flink.table.planner.connectors.DynamicSinkUtils.convertSinkToRel(DynamicSinkUtils.java:208)
	at org.apache.flink.table.planner.connectors.DynamicSinkUtils.convertExternalToRel(DynamicSinkUtils.java:151)
	at org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:276)
	at org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translate$1(PlannerBase.scala:194)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
	at scala.collection.Iterator.foreach(Iterator.scala:937)
Sink Schema definition :
Copy code
public static Schema getAttributionJobOutputSchema() {
        return Schema.newBuilder()
                .column("resource", DataTypes.STRING().notNull())
                .column("id", DataTypes.STRING().notNull())
                .column("duration", DataTypes.STRING().notNull())
                .column("duration_type", DataTypes.STRING().notNull())
                .column("attribute", DataTypes.STRING().notNull())
                .column("value", DataTypes.STRING().notNull())
                .column("year", DataTypes.STRING().notNull())
                .column("month", DataTypes.STRING().notNull())
                .column("day", DataTypes.STRING().notNull())
                .column("record_id", DataTypes.CHAR(36).notNull())
                .column("timestamp", DataTypes.TIMESTAMP_LTZ(3).notNull())
                // This is required for Update of DDB
                .column("_c11", DataTypes.TIMESTAMP_LTZ(3).notNull())
                .build();
    }
2 Views
Open in Slack
PreviousNext
Powered by