backtype.storm.tuple.Tuple.getSourceTask()方法的使用及代码示例

x33g5p2x  于2022-01-29 转载在 其他  
字(9.0k)|赞(0)|评价(0)|浏览(72)

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

Tuple.getSourceTask介绍

[英]Gets the id of the task that created this tuple.
[中]获取创建此元组的任务的id。

代码示例

代码示例来源：origin: alibaba/jstorm

static TaskStream fromTuple(Tuple input) {
    return new TaskStream(input.getSourceTask(), input.getSourceGlobalStreamid());
  }
}

代码示例来源：origin: alibaba/jstorm

@Override
public void execute(Tuple tuple) {
  Object obj = tuple.getValue(0);
  long count = tuple.getLong(1);
  int source = tuple.getSourceTask();
  Map<Integer, Long> subCounts = counts.get(obj);
  if (subCounts == null) {
    subCounts = new HashMap<Integer, Long>();
    counts.put(obj, subCounts);
  }
  // Update the current count for this object
  subCounts.put(source, count);
  // Output the sum of all the known counts so for this key
  long sum = 0;
  for (Long val : subCounts.values()) {
    sum += val;
  }
  collector.emit(new Values(obj, sum));
}

代码示例来源：origin: alibaba/jstorm

private void processWatermark(Tuple input) {
  long watermark = ((Watermark) input.getValue(0)).getTimestamp();
  // emit watermark to downstream tasks
  collector.emit(Common.WATERMARK_STREAM_ID, new Values(input.getValue(0)));
  Integer taskId = input.getSourceTask();
  Long taskWatermark = upstreamWatermarks.get(taskId);
  if (taskWatermark == null) {
    upstreamWatermarks.put(taskId, watermark);
  } else if (watermark > taskWatermark) {
    upstreamWatermarks.put(taskId, watermark);
  }
  // todo: needs optimization to update watermark, a bit slow for now
  long minWatermark = Collections.min(upstreamWatermarks.values());
  if (minWatermark > currentWatermark) {
    currentWatermark = minWatermark;
    LOG.debug("Updating current watermark to {}({})",
        currentWatermark, TimeUtils.format((int) (currentWatermark / 1000L)));
  }
  checkEventTimeWindows();
}

代码示例来源：origin: alibaba/mdrill

obj.put("comp", input.getSourceComponent());
obj.put("stream", input.getSourceStreamId());
obj.put("task", input.getSourceTask());
obj.put("tuple", input.getValues());
sendToSubprocess(obj.toString());

代码示例来源：origin: alibaba/mdrill

public byte[] serialize(Tuple tuple) {
  try {
    _outputter.reset();
    WritableUtils.writeVInt(_dataOutputter, tuple.getSourceTask());
    WritableUtils.writeVInt(_dataOutputter, _ids.getStreamId(tuple.getSourceComponent(), tuple.getSourceStreamId()));
    tuple.getMessageId().serialize(_dataOutputter);
    _kryo.serializeInto(tuple.getValues(), _outputter);
    return _outputter.toByteArray();
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}

代码示例来源：origin: alibaba/jstorm

protected void processBatchTuple(Tuple batchEvent) {
  String stream = batchEvent.getSourceStreamId();
  if (stream.equals(TransactionCommon.BARRIER_STREAM_ID)) {
    Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) batchEvent.getValue(0);
    BatchSnapshot snapshot = (BatchSnapshot) val.getSecond().get(0);
    if (currentBatchTracker.receiveBarrier(batchEvent.getSourceTask()))
      currentBatchTracker.expectedTupleCount += snapshot.getTupleCount();
    else
      LOG.warn("Received expired or unexpected barrier={} from task-{}", snapshot, batchEvent.getSourceTask());
    // LOG.debug("Received batch, stream={}, batchId={}, sourceTask={}, values={}", stream, currentBatchTracker.bactchId,
    // batchEvent.getSourceTask(), snapshot);
    LOG.debug("currentBatchTracker={}\n  processingBatches={}\n  PendingBatches={}\n  intraPendingBatches={}", currentBatchTracker, processingBatches,
        batchCache, intraBatchCache);
    /*
     * if (currentBatchTracker.isAllBarriersReceived()) { if (currentBatchTracker.checkFinish()) finishCurrentBatch(); else
     * outputCollector.fail(batchEvent); }
     */
  } else {
    for (Object value : batchEvent.getValues()) {
      Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) value;
      TupleImplExt tuple = new TupleImplExt(topologyContext, val.getSecond(), val.getFirst(), ((TupleImplExt) batchEvent));
      boltExecutor.execute(tuple);
    }
    currentBatchTracker.incrementReceivedCount(batchEvent.getValues().size());
  }
  if (currentBatchTracker.checkFinish()) {
    finishCurrentBatch();
  }
}

代码示例来源：origin: alibaba/mdrill

public IPersistentMap meta() {
  if(_meta==null) {
    _meta = new PersistentArrayMap( new Object[] {
    makeKeyword("stream"), getSourceStreamId(), 
    makeKeyword("component"), getSourceComponent(), 
    makeKeyword("task"), getSourceTask()});
  }
  return _meta;
}

代码示例来源：origin: alibaba/jstorm

tradeCustomer = (TradeCustomer) input.getValue(1);
} catch (Exception e) {
  LOG.error(input.getSourceComponent() + "  " + input.getSourceTask() + " " + input.getSourceStreamId()
      + " target " +  input);
  throw new RuntimeException(e);

代码示例来源：origin: alibaba/jstorm

int sourceTask = input.getSourceTask();
int uptime = (Integer) input.getValue(0);
TaskStatus executorStatus = new TaskStatus();

代码示例来源：origin: alibaba/jstorm

private BoltMsg createBoltMessage(Tuple input, String genId) {
  BoltMsg boltMsg = new BoltMsg();
  boltMsg.setId(genId);
  boltMsg.setComp(input.getSourceComponent());
  boltMsg.setStream(input.getSourceStreamId());
  boltMsg.setTask(input.getSourceTask());
  boltMsg.setTuple(input.getValues());
  return boltMsg;
}

代码示例来源：origin: alibaba/jstorm

public void processTuple(Tuple tuple) {
  int taskId = tuple.getSourceTask();

代码示例来源：origin: alibaba/jstorm

LOG.info("Unknown source stream, " + stream_id + " from task-" + input.getSourceTask());
return;

代码示例来源：origin: alibaba/jstorm

output.writeInt(tuple.getSourceTask(), true);
output.writeInt(_ids.getStreamId(tuple.getSourceComponent(), tuple.getSourceStreamId()), true);

代码示例来源：origin: com.alibaba.jstorm/jstorm-core

static TaskStream fromTuple(Tuple input) {
    return new TaskStream(input.getSourceTask(), input.getSourceGlobalStreamid());
  }
}

代码示例来源：origin: com.n3twork.storm/storm-core

public void execute(Tuple input) {
  if (_exception != null) {
    throw new RuntimeException(_exception);
  }
  //just need an id
  String genId = Long.toString(_rand.nextLong());
  _inputs.put(genId, input);
  try {
    BoltMsg boltMsg = new BoltMsg();
    boltMsg.setId(genId);
    boltMsg.setComp(input.getSourceComponent());
    boltMsg.setStream(input.getSourceStreamId());
    boltMsg.setTask(input.getSourceTask());
    boltMsg.setTuple(input.getValues());
    _pendingWrites.put(boltMsg);
  } catch(InterruptedException e) {
    throw new RuntimeException("Error during multilang processing", e);
  }
}

代码示例来源：origin: com.alibaba.jstorm/jstorm-core

public void process(Tuple input) {
  int sourceTask = input.getSourceTask();
  int uptime = (Integer) input.getValue(0);

代码示例来源：origin: com.alibaba.jstorm/jstorm-core

public void processBatchTuple(Tuple batchEvent) {
  String stream = batchEvent.getSourceStreamId();
  if (stream.equals(TransactionCommon.BARRIER_STREAM_ID)) {
    Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) batchEvent.getValue(0);
    BatchSnapshot snapshot = (BatchSnapshot) val.getSecond().get(1);
    currentBatchTracker.receiveBarrier(batchEvent.getSourceTask());
    currentBatchTracker.expectedTupleCount += snapshot.getTupleCount();
    LOG.debug("Received batch, stream={}, batchGroupId={}, sourceTask={}, values={}", stream, currentBatchTracker.bactchGroupId, batchEvent.getSourceTask(), snapshot);
    LOG.debug("currentBatchTracker={}, processingBatches={}, pendingBatches={}", currentBatchTracker, processingBatches, batchCache);
  } else {
    for (Object value : batchEvent.getValues()) {
      /*List<Object> firstTupleValue = ((Pair<MessageId, List<Object>>) value).getSecond();
      BatchGroupId batchGroupId = (BatchGroupId) firstTupleValue.get(0);
      if (!batchGroupId.equals(currentBatchTracker.bactchGroupId)) {
        LOG.warn("batchgroupid-{} is not equal to the once of current batch tracker-{}!", batchGroupId, currentBatchTracker.bactchGroupId);
      }*/
      Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) value;
      val.getSecond().remove(0);
      TupleImplExt tuple = new TupleImplExt(topologyContext, val.getSecond(), val.getFirst(), ((TupleImplExt) batchEvent));
      boltExecutor.execute(tuple);
    }
    currentBatchTracker.incrementReceivedCount(batchEvent.getValues().size());
  }
  if (currentBatchTracker.checkFinish()) {
    finishCurrentBatch();
  }
}

代码示例来源：origin: mayconbordin/storm-applications

@Override
public String format(Tuple tuple) {
  Fields schema = context.getComponentOutputFields(tuple.getSourceComponent(), tuple.getSourceStreamId());
  String values = "";
  for (int i=0; i<tuple.size(); i++) {
    if (i != 0) values += ", ";
    values += String.format("%s=%s", schema.get(i), tuple.getValue(i));
  }
  
  return String.format(TEMPLATE, tuple.getSourceComponent(), tuple.getSourceTask(),
      tuple.getSourceStreamId(), tuple.getMessageId().toString(), values);
}

代码示例来源：origin: com.n3twork.storm/storm-core

public byte[] serialize(Tuple tuple) {
  try {
    
    _kryoOut.clear();
    _kryoOut.writeInt(tuple.getSourceTask(), true);
    _kryoOut.writeInt(_ids.getStreamId(tuple.getSourceComponent(), tuple.getSourceStreamId()), true);
    tuple.getMessageId().serialize(_kryoOut);
    _kryo.serializeInto(tuple.getValues(), _kryoOut);
    return _kryoOut.toBytes();
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}

代码示例来源：origin: com.alibaba.jstorm/jstorm-core

private BoltMsg createBoltMessage(Tuple input, String genId) {
  BoltMsg boltMsg = new BoltMsg();
  boltMsg.setId(genId);
  boltMsg.setComp(input.getSourceComponent());
  boltMsg.setStream(input.getSourceStreamId());
  boltMsg.setTask(input.getSourceTask());
  boltMsg.setTuple(input.getValues());
  return boltMsg;
}

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

Tuple

相关文章

    查看更多
    微信公众号

    热门标签

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

    最新文章

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