org.apache.samza.system.WatermarkMessage类的使用及代码示例

x33g5p2x  于2022-02-03 转载在 其他  
字(11.7k)|赞(0)|评价(0)|浏览(106)

本文整理了Java中org.apache.samza.system.WatermarkMessage类的一些代码示例，展示了WatermarkMessage类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。WatermarkMessage类的具体详情如下：
包路径：org.apache.samza.system.WatermarkMessage
类名称：WatermarkMessage

WatermarkMessage介绍

[英]The WatermarkMessage is a control message that is sent out to next stage with a watermark timestamp and the task that produces the watermark.
[中]水印消息是一条控制消息，它与水印时间戳和生成水印的任务一起发送到下一阶段。

代码示例

代码示例来源：origin: apache/samza

public static IncomingMessageEnvelope buildWatermarkEnvelope(SystemStreamPartition ssp, long watermark) {
 return new IncomingMessageEnvelope(ssp, null, null, new WatermarkMessage(watermark, null));
}

代码示例来源：origin: apache/samza

/**
 * Update the state upon receiving a watermark message.
 * @param watermarkMessage message of {@link WatermarkMessage}
 * @param ssp system stream partition
 * @return true iff the stream has a new watermark
 */
void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
 WatermarkState state = watermarkStates.get(ssp);
 if (state != null) {
  state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
 } else {
  LOG.error("SSP {} doesn't have watermark states", ssp);
 }
}

代码示例来源：origin: apache/samza

/**
 * Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
 * a new watermark exits.
 * @param watermarkMessage a {@link WatermarkMessage} object
 * @param ssp {@link SystemStreamPartition} that the message is coming from.
 * @param collector message collector
 * @param coordinator task coordinator
 */
public final void aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
  MessageCollector collector, TaskCoordinator coordinator) {
 LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
 watermarkStates.update(watermarkMessage, ssp);
 long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
 if (currentWatermark < watermark) {
  LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());
  if (watermarkMessage.getTaskName() != null) {
   // This is the aggregation task, which already received all the watermark messages from upstream
   // broadcast the watermark to all the peer partitions
   controlMessageSender.broadcastToOtherPartitions(new WatermarkMessage(watermark), ssp, collector);
  }
  // populate the watermark through the dag
  onWatermark(watermark, collector, coordinator);
  // update metrics
  watermarkStates.updateAggregateMetric(ssp, watermark);
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

/**
 * Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
 * a new watermark exits.
 * @param watermarkMessage a {@link WatermarkMessage} object
 * @param ssp {@link SystemStreamPartition} that the message is coming from.
 * @param collector message collector
 * @param coordinator task coordinator
 */
public final void aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
  MessageCollector collector, TaskCoordinator coordinator) {
 LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
 watermarkStates.update(watermarkMessage, ssp);
 long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
 if (currentWatermark < watermark) {
  LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());
  if (watermarkMessage.getTaskName() != null) {
   // This is the aggregation task, which already received all the watermark messages from upstream
   // broadcast the watermark to all the peer partitions
   controlMessageSender.broadcastToOtherPartitions(new WatermarkMessage(watermark), ssp, collector);
  }
  // populate the watermark through the dag
  onWatermark(watermark, collector, coordinator);
  // update metrics
  watermarkStates.updateAggregateMetric(ssp, watermark);
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

/**
 * Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
 * a new watermark exits.
 * @param watermarkMessage a {@link WatermarkMessage} object
 * @param ssp {@link SystemStreamPartition} that the message is coming from.
 * @param collector message collector
 * @param coordinator task coordinator
 */
public final void aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
  MessageCollector collector, TaskCoordinator coordinator) {
 LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
 watermarkStates.update(watermarkMessage, ssp);
 long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
 if (currentWatermark < watermark) {
  LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());
  if (watermarkMessage.getTaskName() != null) {
   // This is the aggregation task, which already received all the watermark messages from upstream
   // broadcast the watermark to all the peer partitions
   controlMessageSender.broadcastToOtherPartitions(new WatermarkMessage(watermark), ssp, collector);
  }
  // populate the watermark through the dag
  onWatermark(watermark, collector, coordinator);
  // update metrics
  watermarkStates.updateAggregateMetric(ssp, watermark);
 }
}

代码示例来源：origin: org.apache.samza/samza-api

public static IncomingMessageEnvelope buildWatermarkEnvelope(SystemStreamPartition ssp, long watermark) {
 return new IncomingMessageEnvelope(ssp, null, null, new WatermarkMessage(watermark, null));
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

/**
 * Update the state upon receiving a watermark message.
 * @param watermarkMessage message of {@link WatermarkMessage}
 * @param ssp system stream partition
 * @return true iff the stream has a new watermark
 */
void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
 WatermarkState state = watermarkStates.get(ssp);
 if (state != null) {
  state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
 } else {
  LOG.error("SSP {} doesn't have watermark states", ssp);
 }
}

代码示例来源：origin: org.apache.samza/samza-core

/**
 * Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
 * a new watermark exits.
 * @param watermarkMessage a {@link WatermarkMessage} object
 * @param ssp {@link SystemStreamPartition} that the message is coming from.
 * @param collector message collector
 * @param coordinator task coordinator
 */
public final void aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
  MessageCollector collector, TaskCoordinator coordinator) {
 LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
 watermarkStates.update(watermarkMessage, ssp);
 long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
 if (currentWatermark < watermark) {
  LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());
  if (watermarkMessage.getTaskName() != null) {
   // This is the aggregation task, which already received all the watermark messages from upstream
   // broadcast the watermark to all the peer partitions
   controlMessageSender.broadcastToOtherPartitions(new WatermarkMessage(watermark), ssp, collector);
  }
  // populate the watermark through the dag
  onWatermark(watermark, collector, coordinator);
  // update metrics
  watermarkStates.updateAggregateMetric(ssp, watermark);
 }
}

代码示例来源：origin: apache/samza

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

/**
 * Update the state upon receiving a watermark message.
 * @param watermarkMessage message of {@link WatermarkMessage}
 * @param ssp system stream partition
 * @return true iff the stream has a new watermark
 */
void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
 WatermarkState state = watermarkStates.get(ssp);
 if (state != null) {
  state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
 } else {
  LOG.error("SSP {} doesn't have watermark states", ssp);
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

/**
 * Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
 * a new watermark exits.
 * @param watermarkMessage a {@link WatermarkMessage} object
 * @param ssp {@link SystemStreamPartition} that the message is coming from.
 * @param collector message collector
 * @param coordinator task coordinator
 */
public final void aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
  MessageCollector collector, TaskCoordinator coordinator) {
 LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
 watermarkStates.update(watermarkMessage, ssp);
 long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
 if (currentWatermark < watermark) {
  LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());
  if (watermarkMessage.getTaskName() != null) {
   // This is the aggregation task, which already received all the watermark messages from upstream
   // broadcast the watermark to all the peer partitions
   controlMessageSender.broadcastToOtherPartitions(new WatermarkMessage(watermark), ssp, collector);
  }
  // populate the watermark through the dag
  onWatermark(watermark, collector, coordinator);
  // update metrics
  watermarkStates.updateAggregateMetric(ssp, watermark);
 }
}

代码示例来源：origin: apache/samza

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core

/**
 * Update the state upon receiving a watermark message.
 * @param watermarkMessage message of {@link WatermarkMessage}
 * @param ssp system stream partition
 * @return true iff the stream has a new watermark
 */
void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
 WatermarkState state = watermarkStates.get(ssp);
 if (state != null) {
  state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
 } else {
  LOG.error("SSP {} doesn't have watermark states", ssp);
 }
}

代码示例来源：origin: apache/samza

@Test
public void testWatermarkMessageSerde() {
 IntermediateMessageSerde imserde = new IntermediateMessageSerde(new ObjectSerde());
 String taskName = "task-1";
 WatermarkMessage watermark = new WatermarkMessage(System.currentTimeMillis(), taskName);
 byte[] bytes = imserde.toBytes(watermark);
 WatermarkMessage de = (WatermarkMessage) imserde.fromBytes(bytes);
 assertEquals(MessageType.of(de), MessageType.WATERMARK);
 assertEquals(de.getTaskName(), taskName);
 assertTrue(de.getTimestamp() > 0);
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

/**
 * Update the state upon receiving a watermark message.
 * @param watermarkMessage message of {@link WatermarkMessage}
 * @param ssp system stream partition
 * @return true iff the stream has a new watermark
 */
void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
 WatermarkState state = watermarkStates.get(ssp);
 if (state != null) {
  state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
 } else {
  LOG.error("SSP {} doesn't have watermark states", ssp);
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

代码示例来源：origin: org.apache.samza/samza-core

@Override
protected Collection<Void> handleWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
 sendControlMessage(new WatermarkMessage(watermark, taskName), collector);
 return Collections.emptyList();
}

内容来源于网络，如有侵权，请联系作者删除！

WatermarkMessage

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com