Instantly share code, notes, and snippets.

View StreamsJoinWithRepartitioning.java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
View StreamsDSLAndProcessorExample.java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
View ProcessorAPIExample.java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
View StreamsDSLExample.java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
View OptimizingStreams.java
// imports and license left out for clarity
public class OptimizedStreams {
public static void main(String[] args) {
// Details left out for clarity
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> inputStream = builder.stream("inputTopic");
final KStream<String, String> changedKeyStream = inputStream.selectKey((k, v) -> v.substring(0,5));
View punctuateMethodForProcessor.java
// process call left out for clarity; it places airline data
// in a list, storing in state store by airport code
@Override
public void punctuate(long timestamp) {
KeyValueIterator<String, List<String>> allFlights = flights.all();
while (allFlights.hasNext()) {
KeyValue<String, List<String>> kv = allFlights.next();
List<String> flightList = kv.value;
String key = kv.key;
if(flightList.size() >= 100) {
View predictionPredictMethod.java
public static String predict(DataRegression dataRegression) {
try (OnlineLogisticRegression logisticRegression = new OnlineLogisticRegression()) {
FlightData flightData = new FlightData(dataRegression.data);
logisticRegression.readFields(new DataInputStream(new ByteArrayInputStream(dataRegression.coefficients)));
double prediction = logisticRegression.classifyScalar(flightData.vector);
String arrivalPrediction = prediction > 0.5 ? "on-time" : "late";
return String.format("%s predicted to be %s", new Flight(dataRegression.data), arrivalPrediction);
} catch (Exception e) {
LOG.error("Problems with predicting " + dataRegression.data, e);
return null;
View mappingValues.java
dataByAirportStream.join(regressionsByAirPortTable,
(k, v) -> k,
DataRegression::new)
.mapValues(Predictor::predict)
View buildingKStreamAndGlobalKTable.java
// configuration and Serde creation left out for clarity
KStream<String, String> dataByAirportStream = builder.stream("raw-airline-data");
GlobalKTable<String, byte[]> regressionsByAirPortTable = builder.globalTable(Serdes.String(),
byteArraySerde,
"onlineRegression-by-airport");
// stream reads raw data joins with coefficients then makes prediction
dataByAirportStream.join(regressionsByAirPortTable,
(k, v) -> k,
DataRegression::new)
View dataByAirportStream.java
dataByAirportStream.join(regressionsByAirPortTable,(k, v) -> k, DataRegression::new)
.mapValues(Predictor::predict)
.filter((k, v) -> v != null)
.peek((k, v) -> System.out.println("Prediction " + v))
.to("predictions");