https://datahubproject.io logo
Join Slack
Powered by
Hi! :wave: Question about stateful ingestion for ...
# troubleshoot
w
Hi! 👋 Question about stateful ingestion for postgres. It should be supported, but it doesnt work for me 😞 I run the pipeline below, then drop the table in postgres, and run it again. The table metadata in UI is not removed. The same happens when we test with bigquery. 
curl http://<datahub-gms-endpoint>/config
is saying 
statefulIngestionCapable: true
so it should be fine we have 
gms v0.8.26
datahub cli v0.8.26.3
Any help would be much appreciated, thank you!
Copy code
pipeline = Pipeline.create(
        # This configuration is analogous to a recipe configuration.
        {
            "source": {
                "type": "postgres",
                "config": {
                    "env": ENV,
                    "host_port": sql_host_port,
                    "database": database,
                    "username": sql_login,
                    "password": sql_password,
                    "include_views": False,
                    "profiling": {
                        "enabled": True
                    },
                    "stateful_ingestion": {
                        "enabled": True,
                        "remove_stale_metadata": True,
                        "state_provider": {
                            "type": "datahub",
                            "config": {"datahub_api": {"server": datahub_host}},
                        },
                    },
                },
            },
            "pipeline_name": "my_postgres_pipeline_1",
            "sink": {
                "type": "datahub-rest",
                "config": {"server": datahub_host},
            },
        }
)
l
cc @gentle-hamburger-31302 @elegant-doctor-86344 ^
h
@witty-painting-90923 Can you share the logs for the metadata ingestion runs ? For ingestion run before dropping the table and the run after dropping the table ?
n
Hey, i work with Ekaterina, she is off this week. i can answer on this one for now. with the exact same configuration but working on bigquery i noticed that it successfully get the latest checkpoint, and commits the new one. but it actually seems that it never compute the difference between the two. We never get the log "Checking for stale entity removal." so i'm assuming that the method "gen_removed_entity_workunits" it is for some reason never called or one of the conditions is not satisfied. (https://github.com/linkedin/datahub/blob/master/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py#L522)
Copy code
[2022-03-01 11:14:05,369] {process_utils.py:143} INFO - INFO:datahub.ingestion.source.state_provider.datahub_ingestion_checkpointing_provider:Querying for the latest ingestion checkpoint for pipelineName:'bigquery_ingestion', platformInstanceId:'bigquery_no_host_port_no_database', job_name:'common_ingest_from_sql_source'
[2022-03-01 11:14:05,386] {process_utils.py:143} INFO - INFO:datahub.ingestion.source.state_provider.datahub_ingestion_checkpointing_provider:The last committed ingestion checkpoint for pipelineName:'bigquery_ingestion', platformInstanceId:'bigquery_no_host_port_no_database', job_name:'common_ingest_from_sql_source' found with start_time: 2022-03-01 11:07:52.115000+00:00 and a bucket duration of None.
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO - --- Logging error ---
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO - Traceback (most recent call last):
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/state/checkpoint.py", line 96, in create_from_checkpoint_aspect
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO -     config_obj = config_class.parse_obj(config_as_dict)
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO -   File "pydantic/main.py", line 511, in pydantic.main.BaseModel.parse_obj
[2022-03-01 11:14:05,391] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/sql/bigquery.py", line 253, in __init__
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -     super().__init__(**data)
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "pydantic/main.py", line 329, in pydantic.main.BaseModel.__init__
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "pydantic/main.py", line 1022, in pydantic.main.validate_model
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "pydantic/fields.py", line 847, in pydantic.fields.ModelField.validate
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "pydantic/fields.py", line 1118, in pydantic.fields.ModelField._apply_validators
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "pydantic/class_validators.py", line 280, in pydantic.class_validators._generic_validator_cls.lambda3
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/sql/bigquery.py", line 273, in bigquery_doesnt_need_platform_instance
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -     "BigQuery project ids are globally unique. You do not need to specify a platform instance."
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO - datahub.configuration.common.ConfigurationError: BigQuery project ids are globally unique. You do not need to specify a platform instance.
[2022-03-01 11:14:05,392] {process_utils.py:143} INFO -
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO - During handling of the above exception, another exception occurred:
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO - Traceback (most recent call last):
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -   File "/usr/local/lib/python3.6/logging/__init__.py", line 994, in emit
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -     msg = self.format(record)
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -   File "/usr/local/lib/python3.6/logging/__init__.py", line 840, in format
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -     return fmt.format(record)
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -   File "/usr/local/lib/python3.6/logging/__init__.py", line 577, in format
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -     record.message = record.getMessage()
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -   File "/usr/local/lib/python3.6/logging/__init__.py", line 338, in getMessage
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO -     msg = msg % self.args
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO - TypeError: not all arguments converted during string formatting
[2022-03-01 11:14:05,393] {process_utils.py:143} INFO - Call stack:
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/script.py", line 46, in <module>
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -     res = read_and_execute_datahub_recipe(*arg_dict["args"], **arg_dict["kwargs"])
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/script.py", line 43, in read_and_execute_datahub_recipe
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -     pipeline.run()
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/run/pipeline.py", line 182, in run
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -     self.source.get_workunits(), 10 if self.preview_mode else None
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/sql/bigquery.py", line 655, in get_workunits
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -     for wu in super().get_workunits():
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/sql/sql_common.py", line 675, in get_workunits
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -     yield from self.gen_removed_entity_workunits()
[2022-03-01 11:14:05,394] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/sql/sql_common.py", line 524, in gen_removed_entity_workunits
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -     self.get_default_ingestion_job_id(), BaseSQLAlchemyCheckpointState
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/state/stateful_ingestion_base.py", line 195, in get_last_checkpoint
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -     job_id, checkpoint_state_class
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/state/stateful_ingestion_base.py", line 179, in _get_last_checkpoint
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -     state_class=checkpoint_state_class,
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -   File "/tmp/venvnpwplmm7/lib/python3.6/site-packages/datahub/ingestion/source/state/checkpoint.py", line 100, in create_from_checkpoint_aspect
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO -     "Failed to construct checkpoint's config from checkpoint aspect.", e
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO - Message: "Failed to construct checkpoint's config from checkpoint aspect."
[2022-03-01 11:14:05,395] {process_utils.py:143} INFO - Arguments: (ConfigurationError('BigQuery project ids are globally unique. You do not need to specify a platform instance.',),)
[2022-03-01 11:14:05,396] {process_utils.py:143} INFO - INFO:datahub.ingestion.run.pipeline:Processing commit request for DatahubIngestionCheckpointingProvider. Commit policy = CommitPolicy.ON_NO_ERRORS, has_errors=False, has_warnings=False
[2022-03-01 11:14:05,396] {process_utils.py:143} INFO - INFO:datahub.ingestion.source.state_provider.datahub_ingestion_checkpointing_provider:Committing ingestion checkpoint for pipeline:'bigquery_ingestion',instance:'bigquery_no_host_port_no_database', job:'common_ingest_from_sql_source'
[2022-03-01 11:14:05,402] {process_utils.py:143} INFO - INFO:datahub.ingestion.source.state_provider.datahub_ingestion_checkpointing_provider:Committed ingestion checkpoint for pipeline:'bigquery_ingestion',instance:'bigquery_no_host_port_no_database', job:'common_ingest_from_sql_source'
[2022-03-01 11:14:05,402] {process_utils.py:143} INFO - INFO:datahub.ingestion.run.pipeline:Successfully committed changes for DatahubIngestionCheckpointingProvider.
this one is the recipe for bigquery also:
Copy code
pipeline_name: "bigquery_ingestion"
source:
  type: bigquery
  config:
    project_id: project_id
    credential:
      project_id: "${GOOGLE_CREDENTIALS_PROJECT_ID}"
      private_key_id: "${GOOGLE_CREDENTIALS_PRIVATE_KEY_ID}"
      private_key: "${GOOGLE_CREDENTIALS_PRIVATE_KEY}"
      client_email: "${GOOGLE_CREDENTIALS_CLIENT_EMAIL}"
      client_id: "${GOOGLE_CREDENTIALS_CLIENT_ID}"
    include_table_lineage: False
    stateful_ingestion:
      enabled: True
      remove_stale_metadata: True
      state_provider:
        type: "datahub"
        config:
          datahub_api:
            server: ${DATAHUB_GMS_ENDPOINT}
sink:
  type: datahub-rest
  config:
    server: ${DATAHUB_GMS_ENDPOINT}
transformers:
  - type: "add_dataset_properties"
    config:
      add_properties_resolver_class: "common_tasks.datahub.custom_transformers.bq_property_resolver.BQPropertiesResolver"
g
Hi @numerous-application-54063 / @witty-painting-90923 I ran the same recipe on my setup and stateful ingestion is working fine with the Postgres plugin. Please find the detail of my system OS: Ubuntu 20.04 data-hub cli: acryl-datahub, version 0.8.26.3 gms: v0.8.26 Ingestion python code is available at: https://gist.github.com/mohdsiddique/e02268c501dbb396bef694e78bfcc075 Please edit the variables (DATAHUB_GMS_ENDPOINT, HOST_PORT, DB_NAME, USERNAME, PASSWORD) in python code as per your system configuration. Sometimes because of browser cache, it looks like stateful ingestion is not working, However, if we navigate to the dashboard, refresh the browser and thereafter navigate to the dataset, then we can see the updated views. You can try the below steps on your setup to see the state of the dataset entity 1. create a dummy database let’s say foo 2. create two tables t1 and t2 in the foo database 3. run the stateful ingestion 4. Open the GMS graphql endpoint http//&lt;DATAHUB GMS ENDPOINT&gt;9002/api/graphiql 5. Paste below query in the top-left panel
Copy code
{
  search(input: {type: DATASET, query: "*", start: 0, count: 10}) {
    start
    count
    total
    searchResults {
    entity {
      urn
      type
      ... on Dataset {
      name
      status {
        removed
      }
      }
    }
    }
  }
  }
6. Click on play button (It is above the top-left panel, first button) 7. You will see output in right panel 8. Copy the URN of table t1. Consider the URN is urnlidataset: (urnlidataPlatform:postgres,foo.public.t1,DEV) 9. Go to postgres database foo 10. Delete the table t1 11. Run the ingestion again 12. Go to GMS graphql endpoint http//&lt;DATAHUB GMS ENDPOINT&gt;9002/api/graphiql 13. Paste below query in top-left panel. Replace the urnlidataset:(urnlidataPlatform:postgres,foo.public.t1,DEV) by urn value you have captured in step 8.
Copy code
{
  dataset(urn: "urn:li:dataset:(urn:li:dataPlatform:postgres,foo.public.t1,DEV)") {
    urn
    type
    status {
    removed
    }
  }
  }
14. Click on the play button, you can see the removed value is set to true. cc: @elegant-doctor-86344. @loud-island-88694
n
Actually on our side is not failing, it logs the exception but in the source code is catched. i think platform_instance is needed for stateful ingestion so in the connector is fixed to the value "bigquery_no_host_port_no_database". but even if the exception is not raised and the ingestion is completed on our side it seems that the connector doesn't compute the difference with previous checkpoint.
b
but with the exception it cannot recreate the state from checkpoint. So its not possible to compute diff. Your first example was with postgres the second bigquery. Is the issue on both platforms?
n
oh i see what you mean now. Stateful ingestion is not working for us on both postgres and bigquery. but we get the exception mentioned only in bigquery. postgres runs without exception but still don't compute the diff
b
Hi folks - thanks for the details. We are taking a look
❤️ 1
l
@elegant-doctor-86344 @gentle-hamburger-31302 Looks like there are two separate issues: 1) checkpointing is probably not working properly on BigQuery 2) Issue with Postgres - might be better to setup a zoom call with @numerous-application-54063 to debug this more
b
For the first one I already created an bug ticket if you want to track it there https://github.com/linkedin/datahub/issues/4288
n
@loud-island-88694 @gentle-hamburger-31302 i was able to debug postgres now. Actually it was not committing at all because we were having an error reading from tables in information_schema. Then since the commit policy is on no errors, the commit was discarded. The strange thing is that information_schema should be denied by default for postgres source, but in our case it was reading it the same, until i explicitly denied it in the recipe using:
Copy code
"schema_pattern": {
    "deny": ["information_schema"]
}
So i can confirm that postgres is working, and bigquery is a separate issue. my bad, i was missing some logs. Thank you for the support on this one, i will look after for updates on the bigquery issue.
g
Thanks @numerous-application-54063, I will get back to you on bigquery issue.
h
The PR with the fix for the checkpoint reconstruction in BigQuery is out: https://github.com/linkedin/datahub/pull/4295.
🚀 2
❤️ 2
2 Views
Open in Slack
PreviousNext
Powered by