isaacmg

import java.io.File;
import java.util.*;
import java.util.Comparator;
/**
 * Created by isaac on 7/12/17.
 */
public class AirlineBasic {
    public static Node getNodeFromMap(String city2, HashMap<String,Node> nodes){
        Node theNode2 = new Node(city2);
        if(nodes.containsKey(city2)){

isaacmg

//update function 
function update(message1, caller,flowInfo) {  
  if(caller==0){
    var message1 = JSON.parse(message1.value) 
  }
  var tx = message1.some_text;
  var colorS = color(message1.height); 
  svg.append("circle")
        .attr("cx", function(d) {
            return projection([message1.value.timeSeries[0].sourceInfo.geoLocation.geogLocation.longitude, message1.value.timeSeries[0].sourceInfo.geoLocation.geogLocation.latitude])[0];        })

isaacmg

//require everything
var app = require('express')();
var http = require('http').Server(app);
var io = require('socket.io')(http);
//Basic NodeJS stuff 
app.get('/', function(req, res){
  res.sendFile(__dirname + '/index.html');
});
//Call SocketIO with the message from Kafka 
function callSockets(io, message){

isaacmg

List<String> jsonString = getDataString(s);
KafkaProducer<String,String> producer = new KafkaProducer<String,String>(props);
for(String message:jsonString)
    {
    ProducerRecord<String,String> producerRecord = new ProducerRecord<String,String>("test", "a river", message);
    producer.send(producerRecord);
    }
producer.close();

isaacmg

from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime
import os
import sys

args = {
  'owner': 'airflow'
  , 'start_date': datetime(2017, 1, 27)
  , 'provide_context': True

isaacmg

Start zookeeper:

zkserver

Start Kafka

cd (path to kafka) /kafka_2.11-0.10.1.1/

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test .\bin\windows\kafka-server-start.bat .\config\server.properties

Now create producers/consumers

cd bin/windows kafka-console-producer.bat --broker-list localhost:9092 --topic test