Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update example to use LocalDagRunner #25

Merged
merged 5 commits into from
Jul 9, 2021
Merged

Update example to use LocalDagRunner #25

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
10c562c
Updated to LocalDagRunner
Nirzu97 Jul 9, 2021
2ef400b
updated LocalDagRunner
Nirzu97 Jul 9, 2021
882ac26
Update taxi_example_local.py
Nirzu97 Jul 9, 2021
ccc57d2
Update
Nirzu97 Jul 9, 2021
4245af3
Updates
Nirzu97 Jul 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions example/taxi_example_local.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,12 @@
import urllib
import os
from typing import Text
import tfx
import absl
import tfx
from tfx.components import CsvExampleGen, StatisticsGen, SchemaGen
from tfx.orchestration import metadata
from tfx.orchestration import pipeline
from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner
from tfx.orchestration.local import local_dag_runner

sys.path.append(".")
from schemacomponent.component import component
Expand Down Expand Up @@ -66,7 +66,7 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,

# modifies infered schema with use of udf `schema_fn` defined in module file
schema_curation = component.SchemaCuration(schema=schema_gen.outputs['schema'],
module_file='schemacomponent\example\module_file.py')
module_file=os.path.join('schemacomponent','example','module_file.py')


return pipeline.Pipeline(
Expand All @@ -82,9 +82,9 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,
# $python taxi_pipeline_hello.py
if __name__ == '__main__':
absl.logging.set_verbosity(absl.logging.INFO)
BeamDagRunner().run(
local_dag_runner.LocalDagRunner().run(
_create_pipeline(
pipeline_name=_pipeline_name,
pipeline_root=_pipeline_root,
data_root=_data_root,
metadata_path=_metadata_path))
metadata_path=_metadata_path))