GEOFBOT

function flink-debug
    set -lx JVM_ARGS -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005
    eval $argv[1..-1]
end

GEOFBOT

# portions based on slap.py https://github.com/hexchat/hexchat-addons/blob/master/python/slap/slap.py
from __future__ import print_function

import hexchat

__module_name__ = "Shrug"
__module_version__ = "1.1"
__module_description__ = "Shrugs with /SHRUG [target]"
__author__ = 'Geoffrey Mon @GEOFBOT'

GEOFBOT

set nocompatible              " be iMproved, required
filetype off                  " required
set shell=/bin/bash           " Fish shell

" set the runtime path to include Vundle and initialize
set rtp+=~/.vim/bundle/Vundle.vim
call vundle#begin()

set number

GEOFBOT

NOTE: HDFS is required for Flink's DistributedCache which distributes Python plans to worker nodes. We use BlueData Hadoop CDH nodes. Remember to make sure you aren't using env.execute(local=True) in your Python plans!

On the master node:

0. Install git and other useful things that we like

   sudo yum install git bzip2 -y
   

GEOFBOT

package org.apache.flink;

import org.apache.commons.io.Charsets;
import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;

GEOFBOT

# modified from http://www.willmcginnis.com/2015/11/08/getting-started-with-python-and-apache-flink/

from flink.plan.Environment import get_environment
from flink.plan.Constants import INT, STRING, WriteMode
from flink.functions.GroupReduceFunction import GroupReduceFunction

class Adder(GroupReduceFunction):
    def reduce(self, iterator, collector):
        count, word = iterator.next()
        count += sum([x[0] for x in iterator])

GEOFBOT

On each node:

Set up packages and install Hadoop:

#!/bin/bash

sudo yum install java-1.8.0-openjdk-devel wget git bzip2 -y
echo export JAVA_HOME=/usr/lib/jvm/java >> ~/.bashrc
source ~/.bashrc

GEOFBOT

[Unit]
Description=Apache Flink
           
[Service]
Type=forking
User=ec2-user
ExecStart=/home/ec2-user/flink/bin/start-cluster.sh
ExecStop=-/home/ec2-user/flink/bin/stop-cluster.sh
Restart=always
           

GEOFBOT

# Barebones test file to check for issues
import math

from flink.functions.Aggregation import Sum
from flink.functions.GroupReduceFunction import GroupReduceFunction
from flink.plan.Environment import get_environment


class NormalizeVectorGroupReducer(GroupReduceFunction):
    """

GEOFBOT

#!/bin/bash
# Runs after installation of included Flink
set -e
cd ~
sudo rm /usr/lib/flink/lib/flink-* # make sure we don't have two versions of jars
sudo tar -xzf flinkerations-emr.tgz -C /usr/lib/
rm flinkerations-emr.tgz

# Copy over EMRFS jars to Flink lib path
sudo cp /usr/share/aws/emr/s3-dist-cp/lib/*.jar /usr/lib/flink/lib/