本文整理汇总了Java中org.apache.samza.task.TaskCoordinator类的典型用法代码示例。如果您正苦于以下问题:Java TaskCoordinator类的具体用法?Java TaskCoordinator怎么用?Java TaskCoordinator使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
TaskCoordinator类属于org.apache.samza.task包,在下文中一共展示了TaskCoordinator类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: process
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
Map<String, Object> jsonObject = (Map<String, Object>) envelope.getMessage();
WikipediaFeedEvent event = new WikipediaFeedEvent(jsonObject);
try {
Map<String, Object> parsedJsonObject = parse(event.getRawEvent());
parsedJsonObject.put("channel", event.getChannel());
parsedJsonObject.put("source", event.getSource());
parsedJsonObject.put("time", event.getTime());
collector.send(new OutgoingMessageEnvelope(new SystemStream("kafka", "wikipedia-edits"), parsedJsonObject));
} catch (Exception e) {
System.err.println("Unable to parse line: " + event);
}
}
开发者ID:yoloanalytics,项目名称:bigdata-swamp,代码行数:19,代码来源:WikipediaParserStreamTask.java
示例2: handleMessage
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public Collection<JM> handleMessage(M message, MessageCollector collector, TaskCoordinator coordinator) {
try {
KeyValueStore<K, TimestampedValue<M>> thisState = thisPartialJoinFn.getState();
KeyValueStore<K, TimestampedValue<OM>> otherState = otherPartialJoinFn.getState();
K key = thisPartialJoinFn.getKey(message);
thisState.put(key, new TimestampedValue<>(message, clock.currentTimeMillis()));
TimestampedValue<OM> otherMessage = otherState.get(key);
long now = clock.currentTimeMillis();
if (otherMessage != null && otherMessage.getTimestamp() > now - ttlMs) {
JM joinResult = thisPartialJoinFn.apply(message, otherMessage.getValue());
return Collections.singletonList(joinResult);
}
} catch (Exception e) {
throw new SamzaException("Error handling message in PartialJoinOperatorImpl " + getOpImplId(), e);
}
return Collections.emptyList();
}
开发者ID:apache,项目名称:samza,代码行数:21,代码来源:PartialJoinOperatorImpl.java
示例3: handleTimer
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public Collection<WindowPane<K, Object>> handleTimer(MessageCollector collector, TaskCoordinator coordinator) {
LOG.trace("Processing timer.");
List<WindowPane<K, Object>> results = new ArrayList<>();
List<TriggerKey<K>> keys = triggerScheduler.runPendingCallbacks();
for (TriggerKey<K> key : keys) {
TriggerImplHandler triggerImplHandler = triggers.get(key);
if (triggerImplHandler != null) {
Optional<WindowPane<K, Object>> maybeTriggeredPane = triggerImplHandler.onTimer(key, collector, coordinator);
maybeTriggeredPane.ifPresent(results::add);
}
}
LOG.trace("Triggered panes: " + results.size());
return results;
}
开发者ID:apache,项目名称:samza,代码行数:17,代码来源:WindowOperatorImpl.java
示例4: onMessage
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
public Optional<WindowPane<K, Object>> onMessage(TriggerKey<K> triggerKey, M message,
MessageCollector collector, TaskCoordinator coordinator) {
if (!isCancelled) {
LOG.trace("Forwarding callbacks for {}", message);
impl.onMessage(message, triggerScheduler);
if (impl.shouldFire()) {
// repeating trigger can trigger multiple times, So, clear the state to allow future triggerings.
if (impl instanceof RepeatingTriggerImpl) {
((RepeatingTriggerImpl<M, K>) impl).clear();
}
return onTriggerFired(triggerKey, collector, coordinator);
}
}
return Optional.empty();
}
开发者ID:apache,项目名称:samza,代码行数:17,代码来源:WindowOperatorImpl.java
示例5: aggregateEndOfStream
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
/**
* Aggregate {@link EndOfStreamMessage} from each ssp of the stream.
* Invoke onEndOfStream() if the stream reaches the end.
* @param eos {@link EndOfStreamMessage} object
* @param ssp system stream partition
* @param collector message collector
* @param coordinator task coordinator
*/
public final void aggregateEndOfStream(EndOfStreamMessage eos, SystemStreamPartition ssp, MessageCollector collector,
TaskCoordinator coordinator) {
LOG.info("Received end-of-stream message from task {} in {}", eos.getTaskName(), ssp);
eosStates.update(eos, ssp);
SystemStream stream = ssp.getSystemStream();
if (eosStates.isEndOfStream(stream)) {
LOG.info("Input {} reaches the end for task {}", stream.toString(), taskName.getTaskName());
onEndOfStream(collector, coordinator);
if (eosStates.allEndOfStream()) {
// all inputs have been end-of-stream, shut down the task
LOG.info("All input streams have reached the end for task {}", taskName.getTaskName());
coordinator.commit(TaskCoordinator.RequestScope.CURRENT_TASK);
coordinator.shutdown(TaskCoordinator.RequestScope.CURRENT_TASK);
}
}
}
开发者ID:apache,项目名称:samza,代码行数:27,代码来源:OperatorImpl.java
示例6: testStreamOperator
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testStreamOperator() {
StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> mockOp = mock(StreamOperatorSpec.class);
FlatMapFunction<TestMessageEnvelope, TestOutputMessageEnvelope> txfmFn = mock(FlatMapFunction.class);
when(mockOp.getTransformFn()).thenReturn(txfmFn);
Config mockConfig = mock(Config.class);
TaskContext mockContext = mock(TaskContext.class);
StreamOperatorImpl<TestMessageEnvelope, TestOutputMessageEnvelope> opImpl =
new StreamOperatorImpl<>(mockOp, mockConfig, mockContext);
TestMessageEnvelope inMsg = mock(TestMessageEnvelope.class);
Collection<TestOutputMessageEnvelope> mockOutputs = mock(Collection.class);
when(txfmFn.apply(inMsg)).thenReturn(mockOutputs);
MessageCollector mockCollector = mock(MessageCollector.class);
TaskCoordinator mockCoordinator = mock(TaskCoordinator.class);
Collection<TestOutputMessageEnvelope> results = opImpl
.handleMessage(inMsg, mockCollector, mockCoordinator);
verify(txfmFn, times(1)).apply(inMsg);
assertEquals(results, mockOutputs);
}
开发者ID:apache,项目名称:samza,代码行数:21,代码来源:TestStreamOperatorImpl.java
示例7: testSinkOperatorClose
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Test
public void testSinkOperatorClose() {
TestOutputMessageEnvelope mockMsg = mock(TestOutputMessageEnvelope.class);
MessageCollector mockCollector = mock(MessageCollector.class);
TaskCoordinator mockCoordinator = mock(TaskCoordinator.class);
SinkFunction<TestOutputMessageEnvelope> sinkFn = mock(SinkFunction.class);
SinkOperatorImpl<TestOutputMessageEnvelope> sinkImpl = createSinkOperator(sinkFn);
sinkImpl.handleMessage(mockMsg, mockCollector, mockCoordinator);
verify(sinkFn, times(1)).apply(mockMsg, mockCollector, mockCoordinator);
// ensure that close is not called yet
verify(sinkFn, times(0)).close();
sinkImpl.handleClose();
// ensure that close is called once from handleClose()
verify(sinkFn, times(1)).close();
}
开发者ID:apache,项目名称:samza,代码行数:19,代码来源:TestSinkOperatorImpl.java
示例8: testOnMessageUpdatesMetrics
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Test
public void testOnMessageUpdatesMetrics() {
TaskContextImpl mockTaskContext = mock(TaskContextImpl.class);
ReadableMetricsRegistry mockMetricsRegistry = mock(ReadableMetricsRegistry.class);
when(mockTaskContext.getMetricsRegistry()).thenReturn(mockMetricsRegistry);
Counter mockCounter = mock(Counter.class);
Timer mockTimer = mock(Timer.class);
when(mockMetricsRegistry.newCounter(anyString(), anyString())).thenReturn(mockCounter);
when(mockMetricsRegistry.newTimer(anyString(), anyString())).thenReturn(mockTimer);
Object mockTestOpImplOutput = mock(Object.class);
OperatorImpl<Object, Object> opImpl = new TestOpImpl(mockTestOpImplOutput);
opImpl.init(mock(Config.class), mockTaskContext);
// send a message to this operator
MessageCollector mockCollector = mock(MessageCollector.class);
TaskCoordinator mockCoordinator = mock(TaskCoordinator.class);
opImpl.onMessage(mock(Object.class), mockCollector, mockCoordinator);
// verify that it updates message count and timer metrics
verify(mockCounter, times(1)).inc();
verify(mockTimer, times(1)).update(anyLong());
}
开发者ID:apache,项目名称:samza,代码行数:24,代码来源:TestOperatorImpl.java
示例9: testOnTimerUpdatesMetrics
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Test
public void testOnTimerUpdatesMetrics() {
TaskContextImpl mockTaskContext = mock(TaskContextImpl.class);
ReadableMetricsRegistry mockMetricsRegistry = mock(ReadableMetricsRegistry.class);
when(mockTaskContext.getMetricsRegistry()).thenReturn(mockMetricsRegistry);
Counter mockMessageCounter = mock(Counter.class);
Timer mockTimer = mock(Timer.class);
when(mockMetricsRegistry.newCounter(anyString(), anyString())).thenReturn(mockMessageCounter);
when(mockMetricsRegistry.newTimer(anyString(), anyString())).thenReturn(mockTimer);
Object mockTestOpImplOutput = mock(Object.class);
OperatorImpl<Object, Object> opImpl = new TestOpImpl(mockTestOpImplOutput);
opImpl.init(mock(Config.class), mockTaskContext);
// send a message to this operator
MessageCollector mockCollector = mock(MessageCollector.class);
TaskCoordinator mockCoordinator = mock(TaskCoordinator.class);
opImpl.onTimer(mockCollector, mockCoordinator);
// verify that it updates metrics
verify(mockMessageCounter, times(0)).inc();
verify(mockTimer, times(1)).update(anyLong());
}
开发者ID:apache,项目名称:samza,代码行数:24,代码来源:TestOperatorImpl.java
示例10: process
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
if (envelope.getSystemStreamPartition().getStream().equals("epoch")) {
int newEpoch = Integer.parseInt((String) envelope.getMessage());
logger.info("New epoch in message - " + newEpoch);
Integer epoch = getInt(EPOCH);
if (epoch == null || newEpoch == epoch)
return;
if (newEpoch < epoch)
throw new IllegalArgumentException("Got new epoch " + newEpoch + " which is less than current epoch " + epoch);
// it's a new era, reset current epoch and count
logger.info("Epoch: " + newEpoch);
this.state.put(EPOCH, Integer.toString(newEpoch));
this.state.put(COUNT, "0");
coordinator.commit(RequestScope.ALL_TASKS_IN_CONTAINER);
}
}
开发者ID:apache,项目名称:samza,代码行数:20,代码来源:Emitter.java
示例11: window
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public void window(MessageCollector collector,
TaskCoordinator coordinator) {
KeyValueIterator<String, String> entries = store.all();
while (entries.hasNext()) { // c
Entry<String, String> entry = entries.next();
String key = entry.getKey();
String value = entry.getValue();
if (isTimestampKey(key) && Cart.isAbandoned(value)) { // d
String shopper = extractShopper(key);
String cart = store.get(asCartKey(shopper));
AbandonedCartEvent event =
new AbandonedCartEvent(shopper, cart);
collector.send(new OutgoingMessageEnvelope(
new SystemStream("kafka", "derived-events-ch04"), event)); // e
resetShopper(shopper);
}
}
}
开发者ID:alexanderdean,项目名称:Unified-Log-Processing,代码行数:23,代码来源:AbandonedCartStreamTask.java
示例12: process
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
Map<String, Object> edit = (Map<String, Object>) envelope.getMessage();
Map<String, Boolean> flags = (Map<String, Boolean>) edit.get("flags");
Integer editsAllTime = store.get("count-edits-all-time");
if (editsAllTime == null) editsAllTime = 0;
store.put("count-edits-all-time", editsAllTime + 1);
edits += 1;
byteDiff += (Integer) edit.get("diff-bytes");
boolean newTitle = titles.add((String) edit.get("title"));
for (Map.Entry<String, Boolean> flag : flags.entrySet()) {
if (Boolean.TRUE.equals(flag.getValue())) {
counts.compute(flag.getKey(), (k, v) -> v == null ? 0 : v + 1);
}
}
if (!newTitle) {
repeatEdits.inc();
}
}
开发者ID:apache,项目名称:samza-hello-samza,代码行数:25,代码来源:WikipediaStatsStreamTask.java
示例13: window
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public void window(MessageCollector collector, TaskCoordinator coordinator) {
for (int i = 0; i < 100 && messagesSent < NewsfeedConfig.NUM_FOLLOW_EVENTS; i++, messagesSent++) {
String follower = NewsfeedConfig.randomUser();
String followee = NewsfeedConfig.randomUser();
HashMap<String, Object> message = new HashMap<String, Object>();
message.put("event", "follow");
message.put("follower", follower);
message.put("followee", followee);
message.put("time", NewsfeedConfig.currentDateTime());
collector.send(new OutgoingMessageEnvelope(NewsfeedConfig.FOLLOWS_STREAM, followee, null, message));
}
if (messagesSent % 100000 == 0) {
log.info("Generated " + messagesSent + " follow events");
}
if (messagesSent == NewsfeedConfig.NUM_FOLLOW_EVENTS) {
log.info("Finished generating random follower graph");
coordinator.shutdown(RequestScope.CURRENT_TASK);
}
}
开发者ID:ept,项目名称:newsfeed,代码行数:24,代码来源:GenerateFollowsTask.java
示例14: process
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
Map<String, Object> edit = (Map<String, Object>) envelope.getMessage();
Map<String, Boolean> flags = (Map<String, Boolean>) edit.get("flags");
Integer editsAllTime = store.get("count-edits-all-time");
if (editsAllTime == null) editsAllTime = 0;
store.put("count-edits-all-time", editsAllTime + 1);
edits += 1;
titles.add((String) edit.get("title"));
byteDiff += (Integer) edit.get("diff-bytes");
for (Map.Entry<String, Boolean> flag : flags.entrySet()) {
if (Boolean.TRUE.equals(flag.getValue())) {
Integer count = counts.get(flag.getKey());
if (count == null) {
count = 0;
}
count += 1;
counts.put(flag.getKey(), count);
}
}
}
开发者ID:yoloanalytics,项目名称:bigdata-swamp,代码行数:28,代码来源:WikipediaStatsStreamTask.java
示例15: window
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public void window(MessageCollector collector, TaskCoordinator coordinator) {
counts.put("edits", edits);
counts.put("bytes-added", byteDiff);
counts.put("unique-titles", titles.size());
counts.put("edits-all-time", store.get("count-edits-all-time"));
collector.send(new OutgoingMessageEnvelope(new SystemStream("kafka", "wikipedia-stats"), counts));
// Reset counts after windowing.
edits = 0;
byteDiff = 0;
titles = new HashSet<String>();
counts = new HashMap<String, Integer>();
}
开发者ID:yoloanalytics,项目名称:bigdata-swamp,代码行数:16,代码来源:WikipediaStatsStreamTask.java
示例16: handleMessage
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public Collection<Void> handleMessage(M message, MessageCollector collector,
TaskCoordinator coordinator) {
Object key, value;
if (outputStream.isKeyed()) {
key = ((KV) message).getKey();
value = ((KV) message).getValue();
} else {
key = null;
value = message;
}
collector.send(new OutgoingMessageEnvelope(systemStream, null, key, value));
return Collections.emptyList();
}
开发者ID:apache,项目名称:samza,代码行数:16,代码来源:OutputOperatorImpl.java
示例17: handleEndOfStream
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
protected Collection<WindowPane<K, Object>> handleEndOfStream(MessageCollector collector, TaskCoordinator coordinator) {
List<WindowPane<K, Object>> results = new ArrayList<>();
Set<TriggerKey<K>> triggerKeys = new HashSet<>(triggers.keySet());
for(TriggerKey<K> triggerKey : triggerKeys) {
Optional<WindowPane<K, Object>> triggerResult = onTriggerFired(triggerKey, collector, coordinator);
triggerResult.ifPresent(results::add);
}
return results;
}
开发者ID:apache,项目名称:samza,代码行数:11,代码来源:WindowOperatorImpl.java
示例18: onTimer
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
public Optional<WindowPane<K, Object>> onTimer(TriggerKey<K> key, MessageCollector collector,
TaskCoordinator coordinator) {
if (impl.shouldFire() && !isCancelled) {
LOG.trace("Triggering timer triggers");
// repeating trigger can trigger multiple times, So, clear the trigger to allow future triggerings.
if (impl instanceof RepeatingTriggerImpl) {
((RepeatingTriggerImpl<M, K>) impl).clear();
}
return onTriggerFired(key, collector, coordinator);
}
return Optional.empty();
}
开发者ID:apache,项目名称:samza,代码行数:14,代码来源:WindowOperatorImpl.java
示例19: handleMessage
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public Collection<Void> handleMessage(M message, MessageCollector collector,
TaskCoordinator coordinator) {
K key = keyFunction.apply(message);
V value = valueFunction.apply(message);
Long partitionKey = key == null ? 0L : null;
collector.send(new OutgoingMessageEnvelope(systemStream, partitionKey, key, value));
return Collections.emptyList();
}
开发者ID:apache,项目名称:samza,代码行数:10,代码来源:PartitionByOperatorImpl.java
示例20: handleMessage
import org.apache.samza.task.TaskCoordinator; //导入依赖的package包/类
@Override
public Collection<JM> handleMessage(M message, MessageCollector collector, TaskCoordinator coordinator) {
K key = joinOpSpec.getJoinFn().getMessageKey(message);
Object recordValue = table.get(key);
R record = recordValue != null ? (R) KV.of(key, recordValue) : null;
JM output = joinOpSpec.getJoinFn().apply(message, record);
// The support for inner and outer join will be provided in the jonFn. For inner join, the joinFn might
// return null, when the corresponding record is absent in the table.
return output != null ?
Collections.singletonList(output)
: Collections.emptyList();
}
开发者ID:apache,项目名称:samza,代码行数:14,代码来源:StreamTableJoinOperatorImpl.java
注:本文中的org.apache.samza.task.TaskCoordinator类示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论