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

Jump to bottom

Snowflake Support #5500

Open
turb opened this issue Sep 20, 2024 · 4 comments · May be fixed by #5502
Open

Snowflake Support #5500

turb opened this issue Sep 20, 2024 · 4 comments · May be fixed by #5502
Labels
enhancement New feature or request io

Comments

@turb
Copy link
Contributor

turb commented Sep 20, 2024

Hello here,

Apache Beam has Snowflake support, so it's possible to use it with:

  import import org.apache.beam.sdk.io.snowflake.SnowflakeIO

   val thingMapper = new SnowflakeIO.CsvMapper[Thing] {
       override def mapRow(parts: Array[String]): Thing = Thing(Typed1(parts(0)), Typed2(parts(1)), ...)
    }

    val datasource = SnowflakeIO.DataSourceConfiguration.create()
      .withUsernamePasswordAuth(
        args("snowflake-user"), args("snowflake-password"))
      .withServerName("redacted.snowflakecomputing.com")
      .withDatabase("redacted_db")
      .withRole("redacted_role")
      .withWarehouse("redacted_warehouse")
      .withSchema("redacted_schema")

    val read = SnowflakeIO.read()
      .withDataSourceConfiguration(datasource)
      .fromQuery("SELECT COLUM1, COLUMN2... FROM THINGS")
      .withStagingBucketName("gs://paths") // here for Google Cloud
      .withStorageIntegrationName("redacted_integration")
      .withCsvMapper(thingMapper)
      .withCoder(CoderMaterializer.beam(sc, Thing.coder))

    val things: SCollection[Thing] = sc.customInput("things", read)

However a proper scio integration would be great. I suppose derivating Thing to SnowflakeIO.CsvMapper would need some first a PR in magnolify?

I can work on it.
@RustedBones RustedBones added enhancement New feature or request io labels Sep 20, 2024
@RustedBones
Copy link
Contributor

RustedBones commented Sep 20, 2024 •
edited
Loading

For CSV, we've actually relied on the katan library. It looks to be possible to implement the IO simply with a RowDecoder.

We'd be very happy to see a new contribution from your side!

@turb
Copy link
Contributor Author

turb commented Sep 20, 2024

For CSV, we've actually relied on the katan library. It looks to be possible to implement the IO simply with a RowDecoder.

The parser is implemented on the Beam side, using opencsv: only the downstream mapper can be specified. So it would need a PR on Beam to allow to specify another parser.

@RustedBones
Copy link
Contributor

RustedBones commented Sep 20, 2024 •
edited
Loading

I meant to leverage the decoding part of katan, with smth like

   val thingMapper = new SnowflakeIO.CsvMapper[Thing] {
       override def mapRow(parts: Array[String]): Thing =  implicitly[RowDecoder[Thing]].unsafeDecode(parts.toSeq)
    }

turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
dda5487
turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
40e217c
@turb
Copy link
Contributor Author

turb commented Sep 25, 2024

@RustedBones opened #5502

turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
bb6484c
turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
b53021e
turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
1cf2986
turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
d179ff2
turb added a commit to turb/scio that referenced this issue Sep 25, 2024
@turb
Support Snowflake (spotify#5500)
9f6010d
turb added a commit to turb/scio that referenced this issue Oct 16, 2024
@turb
Support Snowflake (spotify#5500)
ff2fde6
turb added a commit to turb/scio that referenced this issue Oct 16, 2024
@turb
Support Snowflake (spotify#5500)
8c434ed
turb added a commit to turb/scio that referenced this issue Oct 16, 2024
@turb
Support Snowflake (spotify#5500)
74d15ae
turb added a commit to turb/scio that referenced this issue Oct 16, 2024
@turb
Support Snowflake (spotify#5500)
901625d
turb added a commit to turb/scio that referenced this issue Nov 4, 2024
@turb
fixup! Support Snowflake (spotify#5500)
42a4674
turb added a commit to turb/scio that referenced this issue Nov 4, 2024
@turb
fixup! Support Snowflake (spotify#5500)
25c5a76
turb added a commit to turb/scio that referenced this issue Nov 8, 2024
@turb
fixup! Support Snowflake (spotify#5500)
a9b56c4
turb added a commit to turb/scio that referenced this issue Nov 8, 2024
@turb
fixup! Support Snowflake (spotify#5500)
9ede3f3
turb added a commit to turb/scio that referenced this issue Nov 14, 2024
@turb
Support Snowflake (spotify#5500)
4fbd1c8
turb added a commit to turb/scio that referenced this issue Nov 14, 2024
@turb
fixup! Support Snowflake (spotify#5500)
0859b24
turb added a commit to turb/scio that referenced this issue Nov 14, 2024
@turb
fixup! Support Snowflake (spotify#5500)
8b0da46
turb added a commit to turb/scio that referenced this issue Nov 14, 2024
@turb
fixup! Support Snowflake (spotify#5500)
b6c8906
turb added a commit to turb/scio that referenced this issue Nov 14, 2024
@turb
fixup! Support Snowflake (spotify#5500)
48252d2
@RustedBones RustedBones linked a pull request Nov 21, 2024 that will close this issue
Open
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
enhancement New feature or request io
Projects
None yet
Milestone
No milestone
Development

Successfully merging a pull request may close this issue.

Support Snowflake (#5500) turb/scio
2 participants
@turb @RustedBones