Source MSSQL: to any Avro or Parquet destination: failed to convert JSON to Avro
- Dominant language
- Python
- Stars
- 22.1k
- Forks
- 5.3k
- PR merge metrics
- PR metrics pending
Description
## Environment
- **Airbyte version**: 0.35.38-alpha
- **OS Version / Instance**: AWS EC2
- **Deployment**: Docker
- **Source Connector and version**: MSSQL - 0.3.22
- **Destination Connector and version**: Databricks built off commit 5ddef8639a88ca81e570fdd43830c89f1da0c266
- **Severity**: Medium
- **Step where error happened**: Sync job
## Current Behavior
When setting up a sync to use cdc incremental load from MSSQL to Databricks I get a `Failed to convert JSON to Avro` error on a decimal field. However when I run the sync using a full refresh there is no issue. This error occurs on a few table but not all
## Expected Behavior
It should be able to sync the data on either full refresh or incremental modes
## Logs
LOG
```
2022-05-18 04:52:45 destination > tech.allegro.schema.json2avro.converter.AvroConversionException: Failed to convert JSON to Avro: Could not evaluate union, field AmountAhead is expected to be one of these: NULL, DOUBLE. If this is a complex type, check if offending field (path: AmountAhead) adheres to schema: 0.00
2022-05-18 04:52:45 destination > at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:129) ~[converter-1.0.1.jar:?]
2022-05-18 04:52:45 destination > at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:118) ~[converter-1.0.1.jar:?]
2022-05-18 04:52:45 destination > at tech.allegro.schema.json2avro.converter.JsonAvroConverter.convertToGenericDataRecord(JsonAvroConverter.java:95) ~[converter-1.0.1.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.s3.avro.AvroRecordFactory.getAvroRecord(AvroRecordFactory.java:39) ~[io.airbyte.airbyte-integrations.connectors-destination-s3-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.s3.parquet.S3ParquetWriter.write(S3ParquetWriter.java:113) ~[io.airbyte.airbyte-integrations.connectors-destination-s3-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.databricks.DatabricksStreamCopier.write(DatabricksStreamCopier.java:109) ~[io.airbyte.airbyte-integrations.connectors-destination-databricks-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.jdbc.copy.CopyConsumerFactory.lambda$recordWriterFunction$0(CopyConsumerFactory.java:104) ~[io.airbyte.airbyte-integrations.connectors-destination-jdbc-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.record_buffer.InMemoryRecordBufferingStrategy.lambda$flushAll$1(InMemoryRecordBufferingStrategy.java:86) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:54) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.record_buffer.InMemoryRecordBufferingStrategy.flushAll(InMemoryRecordBufferingStrategy.java:82) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.record_buffer.InMemoryRecordBufferingStrategy.addRecord(InMemoryRecordBufferingStrategy.java:65) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.destination.buffered_stream_consumer.BufferedStreamConsumer.acceptTracked(BufferedStreamConsumer.java:137) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:50) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
2022-05-18 04:52:45 destination > at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:194) ~[io.airbyte.airbyte-integrations.bases-base-java-0.38.4-alpha.jar:?]
```
## Steps to Reproduce
1.Create a connection for a cdc enabled table with a column of type decimal(28,2)
2.Attempt to sync table using incremental sync mode
Contributor guide
Assessment
This issue has not been assessed yet.