Hi,
I tried to ingest airbyte database to bigquery...
# replication-troubleshootingh
Hrvoje Piasevoli
11/20/2022, 9:16 PMHi,
I tried to ingest airbyte database to bigquery to test the bigquery denormalized connector as the other variant of the connector is useless for nested data.
The airbyte db is a good testing source as it has tables with jsonb columns (eg actor_catalog.catalog).
The result is not what it is supposed to be as the catalog data is represented as a string of json.
Tweaking the streams config using Octavia and setting the catalog column to be type: object results in error.
When I use the regular bigquery destination from airbyte db and compare to eg zendesk_support I see a difference in _airbyte__raw data column;
all the jsonb column content is represented as string and therefore no additional tables get created for airbyte db tables, opposed to zendesk where the data column contains nested json.
Is there something that can be done to preperly use the bigquery denormalized destination to ingest airbyte db (Postgres source) as true parsed struct/repeatable columns
n
Nataly Merezhuk (Airbyte)
11/21/2022, 7:02 PMHi Hrvoje, let me look into this. Could you please provide more info -
• Airbyte version
• Connector versions
• Full logs
Thanks!
h
Hrvoje Piasevoli
11/21/2022, 9:33 PMSure I am using 0.40.17 deployed in GKE,
source is airbyte's own database in cloud SQL
destination is bigquery-denormalized
source connector: airbyte/source-postgres:1.0.18
destination connector: airbyte/destination-bigquery-denormalized: 1.2.7
There are so many logs as a result of my experimentation with setting the streams json_schema for catalog column, that I am afraid it might be confusing.
Instead, could you please verify if you can ingest the airbyte table actor_catalog using airbyte/destination-bigquery-denormalized and verify that the target table in GBQ has proper definition of catalog column (REPEATED RECORD)
Thank you
Hrvoje Piasevoli
11/21/2022, 9:43 PMHere is some more of my research:
When ingesting airbyte db using regular bighquery destination, the raw table has this content (note the catalog is not a dictionary but a string):
Copy code
{
"catalog_hash": "19f77082",
"catalog": "{\"streams\": [{\"name\": \"actor_catalog\", \"namespace\": \"public\", \"json_schema\": ....}",
"created_at": "2022-10-04T10:51:04.164499Z",
"id": "7c3128d5-2289-42f2-8173-b72593c8c793",
"modified_at": "2022-10-04T10:51:04.164499Z"
} Copy code
{
"team_member_count": 0,
"role_type": 1,
"updated_at": "2022-05-11T14:26:32Z",
"configuration": {
"voice_access": false,
"moderate_forums": false,
"side_conversation_create": true,
...
"organization_editing": false,
"chat_access": false
},
"name": "Light agent",
"description": "Can view and add private comments to tickets",
"created_at": "2021-12-09T22:04:58Z"
} Copy code
2022-11-19 14:08:58 destination > Selected loading method is set to: GCS
2022-11-19 14:08:58 destination > S3 format config: {"format_type":"AVRO","flattening":"No flattening"}
2022-11-19 14:08:58 destination > All tmp files will be removed from GCS when replication is finished
2022-11-19 14:08:58 destination > Creating BigQuery staging message consumer with staging ID 749e91fd-f3a5-471e-9c13-18ddc3783836 at 2022-11-19T14:08:52.948Z
2022-11-19 14:08:58 destination > getSchemaFields : {"type":"object","properties":{"id":{"type":"string"},"catalog":{"type":"string","airbyte_type":"object"},"created_at":{"type":"string","format":"timestamp-micros","airbyte_type":"timestamp_with_timezone"},"modified_at":{"type":"string","format":"timestamp-micros","airbyte_type":"timestamp_with_timezone"},"catalog_hash":{"type":"string"}}} namingResolver io.airbyte.integrations.destination.bigquery.BigQuerySQLNameTransformer@78c1372d
2022-11-19 14:08:58 destination > Airbyte Schema is transformed from {"type":"object","properties":{"id":{"type":"string"},"catalog":{"type":"string","airbyte_type":"object"},"created_at":{"type":"string","format":"timestamp-micros","airbyte_type":"timestamp_with_timezone"},"modified_at":{"type":"string","format":"timestamp-micros","airbyte_type":"timestamp_with_timezone"},"catalog_hash":{"type":"string"}}} to [Field{name=id, type=STRING, mode=null, description=null, policyTags=null}, Field{name=catalog, type=STRING, mode=null, description=null, policyTags=null}, Field{name=created_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}, Field{name=modified_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}, Field{name=catalog_hash, type=STRING, mode=null, description=null, policyTags=null}, Field{name=_airbyte_ab_id, type=STRING, mode=null, description=null, policyTags=null}, Field{name=_airbyte_emitted_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}].
2022-11-19 14:08:58 destination > BigQuery write config: BigQueryWriteConfig[streamName=actor_catalog, namespace=null, datasetId=airbyte, datasetLocation=EU, tmpTableId=GenericData{classInfo=[datasetId, projectId, tableId], {datasetId=airbyte, tableId=_airbyte_tmp_pal_actor_catalog}}, targetTableId=GenericData{classInfo=[datasetId, projectId, tableId], {datasetId=airbyte, tableId=actor_catalog}}, tableSchema=Schema{fields=[Field{name=id, type=STRING, mode=null, description=null, policyTags=null}, Field{name=catalog, type=STRING, mode=null, description=null, policyTags=null}, Field{name=created_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}, Field{name=modified_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}, Field{name=catalog_hash, type=STRING, mode=null, description=null, policyTags=null}, Field{name=_airbyte_ab_id, type=STRING, mode=null, description=null, policyTags=null}, Field{name=_airbyte_emitted_at, type=TIMESTAMP, mode=null, description=null, policyTags=null}]}, syncMode=overwrite, stagedFiles=[]]n
Nataly Merezhuk (Airbyte)
11/28/2022, 8:37 PMThanks for all the research! I don't currently have an answer to this, but I'll bring it up with my team on Wednesday. I wonder - could you try another BigQuery database as a source to see if this issue persists?
8 Views