https://prefect.io logo
Channels
Powered by
f

Farid

02/16/2022, 9:32 PM
Hi Is there anything to consider when saving JSON as a 
Result
using Prefect? I dump the dict objects to 
str
JSONs using 
json.dumps
and then save them using 
S3Result
or 
LocalResult
 and noticed both of them get some extra characters added to the beginning or end of json object:
Copy code
��QXQ >> to the beginning
�. >> to the end
which makes it un-parseable. Saving the same object using 
with open()
on local machine does not bear those extra characters
k

Kevin Kho

02/16/2022, 9:35 PM
I think it’s not just 
json.dumps()
but it’s also serialized as bytes. You can see the code for that here and you can actually make and supply your own serializer too
f

Farid

02/16/2022, 10:28 PM
Are there any examples of this? I simply wanted to load the result json files into snowflake but the encoding is preventing that 😞
k

Kevin Kho

02/16/2022, 11:22 PM
Yeah let me find
f

Farid

02/17/2022, 12:10 AM
Thanks Kevin, you’re a LEGEND! I solved this by just passing the `serializer=JSONSerializer()`to the 
S3Result
Object when initialising
k

Kevin Kho

02/17/2022, 12:13 AM
Ah ok that sounds good!
f

Farid

02/17/2022, 2:14 AM
Do you have an example on how I can inherit and expand a task from the task library? I want to use the output of 
task 1
to use inside the query which is then passed to SnowflakeQuery task. This doesn’t work right now because I think `task 1`'s output is only usuable inside another task not when it’s used to format a query in the flow defintion, ie:
Copy code
with Flow() as flow:
    table_name = task_1()
    SnowflakeQuery(query=f"select * from {table_name};")
k

Kevin Kho

02/17/2022, 2:17 AM
In this specific case, I would just tell you to make an intermediate task and pass table_name to that to format the query
✅ 1
But you can subclass any Task in the Task library and then override the 
run
method with your own
SnowflakeQuery(query=task(lambda table_name: f"select * from {table_name};")(table_name))
f

Farid

02/17/2022, 2:27 AM
I would add an intermediary task but just fyi the lambda method did not work:
Copy code
prefect.exceptions.ClientError: [{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}]
k

Kevin Kho

02/17/2022, 2:29 AM
That’s the first I’ve seen..but it’s ok it’s not as pleasant anyway from a coding perspective even if we fix the lambda cuz you get an ugly name in your DAG like 
lambda x
💯 1
7 Views
Powered by