Add *withAttributes Pubsub I/O #546

samschlegel · 2017-04-15T00:42:01Z

Resolves #535, but by adding new input and output functions, as discussed in #538

Open to renaming. Needs tests.

samschlegel · 2017-04-15T00:43:01Z

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

+      val elementCoder = pipeline.getCoderRegistry.getScalaCoder[T]
+      val outputCoder  = pipeline.getCoderRegistry.getScalaCoder[(T, Map[String, String])]
+      val parseFn = Functions.simpleFn { msg: PubsubMessage =>
+        val element = CoderUtils.decodeFromByteArray(elementCoder, msg.getMessage)


This can throw a CoderException. It looks like in some places Beam catches this and rethrows as a RuntimeException. Not sure what we should do here

It only throws exception for corrupt message right? In that case not much we can do and should be fine to let it escalate.

Yep, only for corrupt message.

samschlegel · 2017-04-15T00:43:18Z

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

+      val elementCoder = pipeline.getCoderRegistry.getScalaCoder[T]
+      val outputCoder  = pipeline.getCoderRegistry.getScalaCoder[(T, Map[String, String])]
+      val parseFn = Functions.simpleFn { msg: PubsubMessage =>
+        val element = CoderUtils.decodeFromByteArray(elementCoder, msg.getMessage)


Same as above re: exceptions.

samschlegel · 2017-04-15T01:05:57Z

How should I add tests to this? Just reuse PubsubIO but pass in (String, Map[String, String]), or create a new TestIO type?

nevillelyh · 2017-04-15T11:35:38Z

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

+      val outputCoder  = pipeline.getCoderRegistry.getScalaCoder[(T, Map[String, String])]
+      val parseFn = Functions.simpleFn { msg: PubsubMessage =>
+        val element = CoderUtils.decodeFromByteArray(elementCoder, msg.getMessage)
+        val attributes = msg.getAttributeMap.asScala.toMap


You can use a private class that extends Map[String, String] to do the lazy wrapping. It's fine since the underlying java Map is never exposed.

https://github.com/spotify/scio/blob/master/scio-core/src/main/scala/com/spotify/scio/values/SideInput.scala#L98

Isn't this a little bit too much for pubsub attributes? I mean if someone is using this methods specifically, they do care about getting attributes. Seems like a unnecessary optimization here?

They do care about getting them, but modifying that map is probably unlikely, so delaying the conversion makes sense.

Seems pretty common to wrap j.u.Map as immutable Map instead of the mutable one from asScala. We can move those from SideInput.scala to a util file and reuse them.

The wrapper in SideInput also converts JIterable to Iterable, so would have to specialize for that anyways, no?

No that's specific to the IterableSideInput case. But wouldn't hurt to pull it into a util file like the regular Map[K, V] wrapper.

Since parseFn is a SimpleFunction<T, U>, which is invariant in U, it would have to leak the wrapper. Is that okay, or is there some way to get around that?

What do mean by leak? parseFn and it's signature doesn't leave this method right? Can you provide a snippet?

The output type of parseFn is what's used as the output type of the read transform:

public Read<T> withAttributes(SimpleFunction<PubsubMessage, T> parseFn)

and thus the element type of the returned SCollection. Would casting the wrapper to an instance of Map[String, String] be the way to go?

nevillelyh · 2017-04-15T11:36:27Z

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

+    * @group input
+    */
+  def pubsubSubscriptionWithAttributes[T: ClassTag](sub: String,
+                                      idLabel: String = null,


Nitpick: align arguments?

nevillelyh · 2017-04-15T11:42:57Z

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

+    } else {
+      val elementCoder = pipeline.getCoderRegistry.getScalaCoder[T]
+      val outputCoder  = pipeline.getCoderRegistry.getScalaCoder[(T, Map[String, String])]
+      val parseFn = Functions.simpleFn { msg: PubsubMessage =>


elementCoder & parseFn are identical in pubsubSubscriptionWithAttributes and pubsubTopicWithAttributes, put them in a helper method? You can even put almost everything in a helper method and parameterize only the gio.PubsubIO.read().subscription(topic) vs gio.PubsubIO.read().topic(topic) part.

nevillelyh · 2017-04-15T11:43:32Z

scio-core/src/main/scala/com/spotify/scio/values/SCollection.scala

+    * Save this SCollection as a Pub/Sub topic using the given map as message attributes.
+    * @group output
+    */
+  def saveAsPubsubWithAttributes[V: ClassTag](topic: String)


I guess you want idLabel and timestamp args here too?

samschlegel · 2017-04-18T20:52:40Z

scio-core/src/main/scala/com/spotify/scio/util/JMapWrapper.scala

+import scala.collection.JavaConverters._
+
+private[scio] object JMapWrapper {
+  def ofMultiMap[A, B](self: JMap[A, JIterable[B]]): Map[A, Iterable[B]] =


ofJIterable? 🤷‍♂️

Either way don't have a strong opinion. We can keep ofMultiMap.

nevillelyh · 2017-04-18T23:35:16Z

LGTM but there's some conflict against master. Can you fix? Thanks.

nevillelyh · 2017-04-19T14:03:55Z

Somehow squash works. Weird.
Anyway thanks for the great work!

jonhpowell · 2017-04-19T22:53:16Z

Appreciate the work, guys!

Would you expect it to be immediately included in an upcoming release? If so, when?

nevillelyh · 2017-04-19T23:08:52Z

Don't have any immediate plan and we're still waiting on a lot of upstream changes like Beam release and scala 2.12 fixes. Can you give the snapshot a try first?

jonhpowell · 2017-04-20T12:53:42Z

Sure...we can try the snapshot.

…

On Apr 19, 2017 4:08 PM, "Neville Li" ***@***.***> wrote: Don't have any immediate plan and we're still waiting on a lot of upstream changes like Beam release and scala 2.12 fixes. Can you give the snapshot a try first? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#546 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AH27RMLtgivJkJGdiFHQ3_sHgPcqtM9oks5rxpQFgaJpZM4M-M60> .

smadigan-dexcom · 2017-04-20T18:44:18Z

Hello, I work with jonopwell and tried to build the current source, it fails with
[error] /Users/smadigan/git/scio/scio-schemas/src/main/scala/com/spotify/scio/avro/AvroUtils.scala:54: not found: type TestRecord [error] def newSpecificRecord(i: Int): TestRecord = [error] ^ [error] /Users/smadigan/git/scio/scio-schemas/src/main/scala/com/spotify/scio/avro/AvroUtils.scala:55: not found: type TestRecord [error] new TestRecord(i, i.toLong, i.toFloat, i.toDouble, true, "hello") [error] ^ [error] two errors found [error] (scio-schemas/compile:compileIncremental) Compilation failed

I tried working around that error and hit further errors.

nevillelyh · 2017-04-20T18:56:26Z

Are you building in IntelliJ? There's a known IntelliJ issue that can be worked around, see #543
We also publish snapshots on every master commit so you don't have to build yourself.

smadigan-dexcom · 2017-04-21T00:24:59Z

I built from the command line, sbt compile
Where are the snapshots published? (Forgive my ignorance, please)

samschlegel · 2017-04-21T00:27:50Z

They're published to the Sonatype Snapshots repo which you can add to your build.sbt with:

 resolvers += Resolver.sonatypeRepo("snapshots")

nevillelyh · 2017-04-21T00:57:59Z

Added this to the wiki FAQ.
https://github.com/spotify/scio/wiki/FAQ#how-do-i-use-the-snapshot-builds

dgouyette · 2017-05-04T08:43:51Z

There is no unit test on saveAsPubsubWithAttributes.
There is one unit test on saveAsPubsub

The signature of saveAsPubsubWithAttributes is difficult to read. An example with a unit test would be a good thing

nevillelyh · 2017-05-04T14:23:00Z

Filed #590, would be great if you already have something and can submit a PR.

samschlegel and others added 3 commits April 14, 2017 17:35

Add Pubsub input and outputs with attribute support

c0a9717

Remove extra space

0941ab2

Remove unnecessary import

ca915f9

samschlegel commented Apr 15, 2017

View reviewed changes

nevillelyh reviewed Apr 15, 2017

View reviewed changes

samschlegel added 5 commits April 17, 2017 11:40

Merge remote-tracking branch 'upstream/master'

6ccfc31

Add idLabel and timestampLabel to saveAsPubsubWithAttributes

99bb6b5

Refactor pubsub inputs to use helper method

73021c1

Pull immutable jmap wrapper out into utils

a64a816

Use JMapWrapper for Pubsub attribute map

09688ab

samschlegel commented Apr 18, 2017

View reviewed changes

nevillelyh merged commit d399eef into spotify:master Apr 19, 2017

nevillelyh mentioned this pull request May 4, 2017

Unit test for saveAsPubsubWithAttributes #590

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add *withAttributes Pubsub I/O #546

Add *withAttributes Pubsub I/O #546

samschlegel commented Apr 15, 2017 •

edited

Loading

samschlegel Apr 15, 2017

nevillelyh Apr 15, 2017

samschlegel Apr 17, 2017

samschlegel Apr 15, 2017

samschlegel commented Apr 15, 2017

nevillelyh Apr 15, 2017

ravwojdyla Apr 17, 2017

samschlegel Apr 17, 2017

nevillelyh Apr 17, 2017

samschlegel Apr 17, 2017

nevillelyh Apr 18, 2017

samschlegel Apr 18, 2017

nevillelyh Apr 18, 2017

samschlegel Apr 18, 2017

nevillelyh Apr 15, 2017

nevillelyh Apr 15, 2017

nevillelyh Apr 15, 2017

samschlegel Apr 18, 2017

nevillelyh Apr 18, 2017

nevillelyh commented Apr 18, 2017

nevillelyh commented Apr 19, 2017

jonhpowell commented Apr 19, 2017

nevillelyh commented Apr 19, 2017

jonhpowell commented Apr 20, 2017 via email

smadigan-dexcom commented Apr 20, 2017

nevillelyh commented Apr 20, 2017

smadigan-dexcom commented Apr 21, 2017

samschlegel commented Apr 21, 2017

nevillelyh commented Apr 21, 2017

dgouyette commented May 4, 2017

nevillelyh commented May 4, 2017

Add *withAttributes Pubsub I/O #546

Add *withAttributes Pubsub I/O #546

Conversation

samschlegel commented Apr 15, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samschlegel commented Apr 15, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nevillelyh commented Apr 18, 2017

nevillelyh commented Apr 19, 2017

jonhpowell commented Apr 19, 2017

nevillelyh commented Apr 19, 2017

jonhpowell commented Apr 20, 2017 via email

smadigan-dexcom commented Apr 20, 2017

nevillelyh commented Apr 20, 2017

smadigan-dexcom commented Apr 21, 2017

samschlegel commented Apr 21, 2017

nevillelyh commented Apr 21, 2017

dgouyette commented May 4, 2017

nevillelyh commented May 4, 2017

samschlegel commented Apr 15, 2017 •

edited

Loading