https://prefect.io
Join Slack
Hi. I have a flow that copies data from S3 to Snowflake. It works fine locally, but when I run it in...
# prefect-server
z
Hi. I have a flow that copies data from S3 to Snowflake. It works fine locally, but when I run it in ECS using Fargate, I'm getting a strange error.
Copy code
Failed to load and execute flow run: ImportError("cannot import name 'List' from 'pip' (/usr/local/lib/python3.7/site-packages/pip/__init__.py)")
this is the flow code.
Copy code
FLOW_NAME = 's3_to_sf_raw'
STORAGE = GitHub(
	repo=REPO,
	path=f"flows/ppi/{FLOW_NAME}.py",
	access_token_secret=PREFECT_GITHUB_ACCESS_TOKEN_NAME,
	ref=GITHUB_BRANCH_NAME
)
RUN_CONFIG = set_run_config(run_config_type=RUN_CONFIG_TYPE)
logger = prefect.context.get('logger')


with Flow(FLOW_NAME, storage=STORAGE, run_config=RUN_CONFIG,) as flow:
	dataset_name = Parameter(PREFECT_DATASET_PARAM_NAME)
	run_env = Parameter(PREFECT_RUN_ENVIRONMENT_PARAM_NAME, default='dev')
	run_env_upper = "DEV"

	sf_user = PrefectSecret(SNOWFLAKE_PREFECT_USERNAME.format(ENVIRONMENT=run_env_upper))
	sf_pwd = PrefectSecret(SNOWFLAKE_PREFECT_PASSWORD.format(ENVIRONMENT=run_env_upper))
	sf_account = PrefectSecret(SNOWFLAKE_ACCOUNT_ID.format(ENVIRONMENT=run_env_upper))

	snowflake_copy = get_dict_kwargs(
		account=sf_account,
		user=sf_user,
		password=sf_pwd,
		database=SNOWFLAKE_RAW_DATABASE,
		schema=SNOWFLAKE_DEFAULT_SCHEMA,
		role=SNOWFLAKE_LOADER_ROLE,
		warehouse=SNOWFLAKE_LOADER_WAREHOUSE,
		table=dataset_name + "_RAW",
        stage=SNOWFLAKE_PROPERTY_RAW_STAGE + "_" + run_env,
		file_format="jsonf"
	)

	copy_run = run_snowflake_copy(snowflake_copy, dataset_name)
	copy_run_out = print_prefect_output(copy_run)


flow.set_reference_tasks([copy_run])
k
You made your own image? Could i see the Dockerfile?
z
very simple
Copy code
FROM prefecthq/prefect:latest

WORKDIR /app

RUN mkdir .dbt

ADD . .
RUN pip install .
k
That is so weird. I guess I would try upgrading pip in the image?
z
Its working now, upgrading did it. Not sure what the issue was
k
me too
z
Another related question. When the copy command runs successfully, but no records are loaded (because theres no new data) the flow is successful. Is there a way to configure this so that it returns a failure or another output, so that we can easily see that no records were loaded?
k
Copy code
@task
def something():
    if cond:
        raise FAIL()
z
How would we check the condition in this case?
k
where FAIL is the singal
The length of the records object?
z
ok ill try that, thanks
3 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by