Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/2940#discussion_r19498949
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala
---
@@ -134,17 +137,32 @@ private[streaming] class ReceiverSupervisorImpl(
optionalMetadata: Option[Any],
optionalBlockId: Option[StreamBlockId]
) {
+ pushAndReportBlock(ByteBufferBlock(bytes), optionalMetadata,
optionalBlockId)
+ }
+
+ /** Store block and report it to driver */
+ def pushAndReportBlock(
+ receivedBlock: ReceivedBlock,
+ optionalMetadata: Option[Any],
+ optionalBlockId: Option[StreamBlockId]
+ ) {
val blockId = optionalBlockId.getOrElse(nextBlockId)
+ val numRecords = receivedBlock match {
+ case ArrayBufferBlock(arrayBuffer) => arrayBuffer.size
+ case _ => -1
+ }
+
val time = System.currentTimeMillis
- blockManager.putBytes(blockId, bytes, storageLevel, tellMaster = true)
- logDebug("Pushed block " + blockId + " in " +
(System.currentTimeMillis - time) + " ms")
- reportPushedBlock(blockId, -1, optionalMetadata)
- }
+ val fileSegmentOption = receivedBlockHandler.storeBlock(blockId,
receivedBlock) match {
--- End diff --
In spark we usually don't add "Option" to variable names that are options.
You just know it from the type. This is actually not even consistent with this
file (optionalMetadata vs fileSegmentOption). I'd prefer to just have it be
named `fileSegment`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
