Run sling from python

[matanper]

I'm using Databricks to run dbt jobs, and I'd like to run a sync after job completion.
I don't think I can run CLI tool from Databricks, is there an API to run sling with python?

[flarco]

Yes, try pip install sling

…

On Wed, May 22, 2024, 5:49 AM matanper ***@***.***> wrote: I'm using Databricks to run dbt jobs, and I'd like to trigger a sync after job completion. I don't think I can run CLI tool from Databricks, is there an API to run sling with python? — Reply to this email directly, view it on GitHub <#299>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AB2QZYSVEVC22S75JGVHD63ZDRLX5AVCNFSM6AAAAABIDFH2KKVHI2DSMVQWIX3LMV43ERDJONRXK43TNFXW4OZWG4YDSNRUHA> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[matanper]

Thanks, I couldn't find how to define connection, is it also possible from python?

[flarco]

You can pass environment variables for your connections. You'll find instructions in the docs.

…

On Wed, May 22, 2024, 6:11 AM matanper ***@***.***> wrote: Thanks, I couldn't find how to define connection, is it also possible from python? — Reply to this email directly, view it on GitHub <#299 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AB2QZYRCNXMADICMREKGGBDZDROMVAVCNFSM6AAAAABIDFH2KKVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4TKMJYHA2DS> . You are receiving this because you commented.Message ID: ***@***.***>

[matanper]

Can you tell me if this is correct?

from urllib.parse import quote_plus

import sling

pg_host = 'x'
pg_password = 'x'

rs_host = 'x'
rs_password = 'x'

env = {
    'SLING_PG': f'postgresql://sling:{quote_plus(pg_password)}@{pg_host}:5432/db',
    'SLING_REDSHIFT': f'redshift://readonly_user:{quote_plus(rs_password)}@{rs_host}:5439/db'
}


source = sling.Source(conn='SLING_REDSHIFT', stream='schema.table',
                      primary_key=['id'], update_key='updated_at')

target = sling.Target(conn='SLING_PG', object='schema.target')

sling.Sling(mode='incremental', source=source, target=target).run(env=env, return_output=True)

I get the following error and not sure why

Exception: 12:54PM INF connecting to source database (redshift)
12:54PM INF connecting to target database (postgres)
12:54PM INF getting checkpoint value
12:54PM INF reading from source database
Sling command failed:
[]

I also tried to run json file it generated directly from the CLI:

❯ sling run -r /var/folders/yd/qj33gs652696qklcq7hnglx00000gn/T/sling-task-e3b3df4b-8167-490e-870c-f42a80a58fd7.json
fatal:
~ Error parsing replication config
did not find 'streams' key

[flarco]

Looks right.
Check out https://github.com/slingdata-io/sling-python/blob/main/sling/sling/__init__.py for source code.
For the config, Try sling run -c /path/to/config.json
You can also try the sling.Replication(), which you can run with sling run -r /path/to/replication.json

[matanper]

I changed to Replication and it seems to work better, but still fails. I tried to run from CLI and I see:

1:39PM INF reading from source database
[1]    56093 killed     sling run -r

Do you think it might be OOM?

[matanper]

It seems like OOM from looking at top, I tried adding source_options=sling.SourceOptions(limit=1_000) but this doesn't solve it. any suggestion?

[flarco]

Ah. How much memory does the machine have?

…

On Thu, May 23, 2024, 8:12 AM matanper ***@***.***> wrote: It seems like OOM from looking at top, I tried adding source_options=sling.SourceOptions(limit=1_000) but this doesn't solve it. any suggestion? — Reply to this email directly, view it on GitHub <#299 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AB2QZYQES5OIBK42F2LQJIDZDXFIHAVCNFSM6AAAAABIDFH2KKVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4TKMZTGMZTM> . You are receiving this because you commented.Message ID: ***@***.***>

[matanper]

My machine has 16GB overall, about ~3GB free
but just to test I tried to do a backfill and gave it a range of data which is empty, it's still getting OOMed

[flarco]

Are you providing props aws_bucket / aws_access_key_id / aws_secret_access_key for redshift? If not, can you try with it?
See https://docs.slingdata.io/connections/database-connections/redshift

[matanper]

Sorry I didn't see it's mandatory 🙏
It made progress but now I'm getting error on timestamp:

pq: invalid input syntax for type timestamp: ""
Sling command failed:
[]

although it should read empty as null as far as I understand

[flarco]

can you set source_options.empty_as_null = true?

[matanper]

Sorry it seems like the cause is using "incremental" mode on an empty table. I changed to "truncate".
Now I'm having permissions issues with my bucket "Error with ListObjectsV2" but I'll manage

[matanper]

I couldn't figure out the problem, I even opened a public bucket:

4:21PM INF unloading from redshift to s3
4:21PM INF execution failed
4:21PM INF ~ Error with ListObjectsV2 for: {
  Bucket: "matan-sling-data-temp",
  Prefix: "temp/stream/1716470507720.csv"
}
RequestCanceled: request context canceled
caused by: context canceled

is there way to see more detailed error?

[flarco]

yes, can you run in debug mode? You can set sling.Replication(..., debug=True) in python.
Or pass in the env var DEBUG=true

[matanper]

Thanks! I got the UNLOAD command from the debug and tried to run it and saw it fails on "LIMIT" not supported inside UNLOAD. dropping the limit from the source solved it

[matanper]

Hi, it looks like the UNLOAD doesn't run with HEADER which leads to the CSVs to be generated without header.
Then it fails to load since it just gives them arbitrary col_x names

2024-05-24 00:53:18 INF ~ Error executing: insert into "target"."table" ("id", "updated_at", "user_id", "col_001", "col_002", "col_003") select "id", "updated_at"::timestamp with time zone as "updated_at", "user_id"::uuid as "user_id", "col_001", "col_002", "col_003" from "target"."table_tmp"

[matanper]

I see it changed in this commit but not sure how to workaround that

[flarco]

I'll have to try to reproduce on my side. Can you open an issue with more details on the error including your replication config?
The log output is useful in that case to give context on the sequence of commands.