cassandra-flambo-save.clj

(ns com.fire.kingdom.flambit
  (:require [flambo.api :as f]
            [flambo.tuple :as ft]
            [clojure.string :as s])
  (:import [com.datastax.spark.connector.japi CassandraJavaUtil]))

; in Cassandra ...
; CREATE KEYSPACE test WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 };
; CREATE TABLE test.words (word text PRIMARY KEY, count int); 

(-> (f/text-file sc "data.txt")
    (f/flat-map (f/fn [l] (s/split l #" ")))
    (f/map-to-pair (f/fn [w] (ft/tuple w 1)))
    (f/reduce-by-key (f/fn [x y] (+ x y)))
    (CassandraJavaUtil/javaFunctions)                                                                                                                                                                                                       
    (.writerBuilder "test" "words" (CassandraJavaUtil/mapTupleToRow (class "foo") (class 1)))                                                                                                                                             
    (.withColumnSelector (CassandraJavaUtil/someColumns (into-array String ["word" "count"])))                                                                                                                                          
    .saveToCassandra)