Uh oh!
There was an error while loading. Please reload this page.
[BEAM-53] Java-only Pubsub sink for streaming. - #171
Conversation
mshields822
commented
Apr 12, 2016
R: @kennknowles |
mshields822
commented
Apr 12, 2016
Note the progression were on is: |
| // ================================================================================ | ||
| /** | ||
| * Number of cores available for publishing. |
There was a problem hiding this comment.
What is a core in this context? My initial instinct is across machines, but it seems to be being used as the sharding factor.
There was a problem hiding this comment.
This is now 'numShards' and needs to be chosen to balance num records batch with pubsub latency. Note that a random long indeed shards but results in most bundles having a single element which kills apiary/grpc quota.
I currently (ie in a pending branch) hard code this in PubsubIO. It is comparable to the heuristics we have baked in for calculating the initial splits for an UnboundedReader.
kennknowles
commented
Apr 14, 2016
Better to have the owner of I/O review this. Also just want to call out here that the follow-ups suggested on #120 should probably go first, but I'll leave that up to Dan. |
mshields822
commented
Apr 14, 2016
ACK working on a pubsub-apiary follow up. Thanks! On Wed, Apr 13, 2016 at 7:16 PM, Kenn Knowles notifications@github.com
|
| elementCounter.addValue(1L); | ||
| byte[] elementBytes = CoderUtils.encodeToByteArray(elementCoder, c.element()); | ||
| long timestampMsSinceEpoch = c.timestamp().getMillis(); | ||
| c.output(KV.of(ThreadLocalRandom.current().nextInt(numCores * SCALE_OUT), |
There was a problem hiding this comment.
Using a random int will allow most systems to choose the sharding to an even more arbitrary degree, and then you can remove numCores
dhalperi
commented
Apr 25, 2016
Looking into this -- looks like you'll have to rebase. I see this code is pre-name-change. |
mshields822
commented
May 4, 2016
PTAL |
| /** | ||
| * Coder for conveying outgoing messages between internal stages. | ||
| */ | ||
| private static final Coder<PubsubClient.OutgoingMessage> CODER = new |
There was a problem hiding this comment.
Noting that this is a case we support quite poorly today: A library author offers a sink, requiring a user to convert to a particular datatype Foo. The library author writes a coder for Foo and would like the user to get this benefit automatically.
Not necessarily directly applicable here, but putting it out there.
There was a problem hiding this comment.
Note this is entirely internal to the Sink / PubsubClient interface. I could put the coder in PubsubClient but since it has no other PCollection/Coder/etc dependencies it felt better leaving it outside.
kennknowles
commented
May 6, 2016
Added a couple initial comments to share the review load. |
| elementCounter.addValue(1L); | ||
| byte[] elementBytes = CoderUtils.encodeToByteArray(elementCoder, c.element()); | ||
| long timestampMsSinceEpoch = c.timestamp().getMillis(); | ||
| c.output(KV.of(ThreadLocalRandom.current().nextInt(numShards), |
There was a problem hiding this comment.
it is preferable (in terms of serialization overhead, which I understand is important for streaming) to make these static inner classes and then pass in constants in constructors. But up to you.
mshields822
commented
May 10, 2016
PTAL. |
| * BLOCKING | ||
| * Send {@code messages} as a batch to Pubsub. | ||
| */ | ||
| private void publishBatch(List<PubsubClient.OutgoingMessage> messages, int bytes) |
There was a problem hiding this comment.
I don't much care, but you write this as both OutgoingMessage and PubsubClient.OutgoingMessage in this file. Probably could pick one and stick with it.
dhalperi
commented
May 11, 2016
Generally looks good to me. Let's sync tomorrow over any remaining comments. |
mshields822
commented
May 11, 2016
PTAL |
| long nowMsSinceEpoch = System.currentTimeMillis(); | ||
| int n = pubsubClient.publish(topic, messages); | ||
| Preconditions.checkState(n == messages.size()); | ||
| checkState(n == messages.size(), "Attempted to publish %d messaged but %d were successful", |
No description provided.