Apache Spark или возвращение блудного пользователя

public static class Map extends Mapper<LongWritable, Text, Text, Text> {    @Override    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {        String s = value.toString();        String[] split = s.split(" ");        context.write(new Text(split[0]), new Text(split[1]));    } }  public static class Reduce extends Reducer<Text, Text, Text, ArrayWritable> {    @Override    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {        HashSet<Text> set = new HashSet<>();        values.forEach(t -> set.add(t));         ArrayWritable array = new ArrayWritable(Text.class);        array.set(set.toArray(new Text[set.size()]));        context.write(key, array);    } }  public static class Run {    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {        Job job = Job.getInstance();        job.setJarByClass(Run.class);        job.setMapperClass(Map.class);        job.setReducerClass(Reduce.class);        job.setOutputKeyClass(Text.class);        job.setOutputValueClass(ArrayWritable.class);        FileInputFormat.addInputPath(job, new Path(args[0]));        FileOutputFormat.setOutputPath(job, new Path(args[1]));        System.exit(job.waitForCompletion(true) ? 0 : 1);    } }  

A = LOAD '/data/input' USING PigStorage(' ') AS (visitor_id:chararray, audience_id:chararray); B = DISTINCT A; C = GROUP B BY visitor_id; D = FOREACH C GENERATE group AS visitor_id, B.audience_id AS audience_id; STORE D INTO '/data/output' USING PigStorage(); 

SparkConf sparkConf = new SparkConf().setAppName("Test"); JavaSparkContext jsc = new JavaSparkContext(sparkConf); jsc.textFile(args[0])        .mapToPair(str -> {            String[] split = str.split(" ");            return new Tuple2<>(split[0], split[1]);        })        .distinct()        .groupByKey()        .saveAsTextFile(args[1]); 

public class StreamUtil {    private static final Function<JavaPairRDD<String, byte[]>, JavaRDD<Event>> eventTransformFunction =            rdd -> rdd.map(t -> Event.parseFromMsgPack(t._2())).filter(e -> e != null);     public static JavaPairReceiverInputDStream<String, byte[]> createStream(JavaStreamingContext jsc, String groupId, Map<String, Integer> topics) {        HashMap prop = new HashMap() {{            put("zookeeper.connect", BaseUtil.KAFKA_ZK_QUORUM);            put("group.id", groupId);        }};        return KafkaUtils.createStream(jsc, String.class, byte[].class, StringDecoder.class, DefaultDecoder.class, prop, topics, StorageLevel.MEMORY_ONLY_SER());    }     public static JavaDStream<Event> getEventsStream(JavaStreamingContext jssc, String groupName, Map<String, Integer> map, int count) {        return getStream(jssc, groupName, map, count, eventTransformFunction);    }     public static <T> JavaDStream<T> getStream(JavaStreamingContext jssc, String groupName, Map<String, Integer> map,                                               Function<JavaPairRDD<String, byte[]>, JavaRDD<T>> transformFunction) {        return createStream(jssc, groupName, map).transform(transformFunction);    }     public static <T> JavaDStream<T> getStream(JavaStreamingContext jssc, String groupName, Map<String, Integer> map, int count,                                               Function<JavaPairRDD<String, byte[]>, JavaRDD<T>> transformFunction) {        if (count < 2) return getStream(jssc, groupName, map, transformFunction);         ArrayList<JavaDStream<T>> list = new ArrayList<>();        for (int i = 0; i < count; i++) {            list.add(getStream(jssc, groupName, map, transformFunction));        }        return jssc.union(list.get(0), list.subList(1, count));    } } 

Для создания потока сообщений нужно передать контекст, необходимые топики и имя группы получателей (jssc, topics и groupId соответственно). Для каждой группы формируется свой сдвиг очереди сообщений по каждому топику. Также можно создавать несколько получателей для распределения нагрузки между серверами. Все преобразования над данными указываются в transformFunction и выполняются в том же потоке, что и получатели.

Создание контекста

   public JavaPairRDD<String, Condition> conditions;    private JavaStreamingContext jssc;    private Map<Object, HyperLogLog> hlls;    public JavaStreamingContext create() {        sparkConf.setAppName("UniversalStreamingBuilder");        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");        sparkConf.set("spark.storage.memoryFraction", "0.125");         jssc = new JavaStreamingContext(sparkConf, batchInterval);        HashMap map = new HashMap();        map.put(topicName, 1); // Kafka topic name and number partitions         JavaDStream<Event> events = StreamUtil.getEventsStream(jssc, groupName, map, numReceivers).repartition(numWorkCores);         updateConditions();         events.foreachRDD(ev -> {            // Compute audiences            JavaPairRDD<String, Object> rawva = conditions.join(ev.keyBy(t -> t.pixelId))                    .mapToPair(t -> t._2())                    .filter(t -> EventActionUtil.checkEvent(t._2(), t._1().condition))                    .mapToPair(t -> new Tuple2<>(t._2().visitorId, t._1().id))                    .distinct()                    .persist(StorageLevel.MEMORY_ONLY_SER())                    .setName("RawVisitorAudience");             // Update HyperLogLog`s            rawva.mapToPair(t -> t.swap()).groupByKey()                    .mapToPair(t -> {                        HyperLogLog hll = new HyperLogLog();                        t._2().forEach(v -> hll.offer(v));                        return new Tuple2<>(t._1(), hll);                    }).collectAsMap().forEach((k, v) -> hlls.merge(k, v, (h1, h2) -> HyperLogLog.merge(h1, h2)));             // Save to Aerospike and HBase            save(rawva);             return null;        });        return jssc;    } 

Здесь, чтобы соединить два (events и conditions) RDD (Resilient Distributed Dataset), используется join по pixel_id. Метод save — фейковый. Это сделано для того, чтобы разгрузить представленный код. На его месте должно находиться несколько преобразований и сохранений.

Запуск

   public void run() {        create();        jssc.start();        long millis = TimeUnit.MINUTES.toMillis(CONDITION_UPDATE_PERIOD_MINUTES);        new Timer(true).schedule(new TimerTask() {            @Override            public void run() {                updateConditions();            }        }, millis, millis);         new Timer(false).scheduleAtFixedRate(new TimerTask() {            @Override            public void run() {                flushHlls();            }        }, new Date(saveHllsStartTime), TimeUnit.MINUTES.toMillis(HLLS_UPDATE_PERIOD_MINUTES));         jssc.awaitTermination();    } 

Вначале создается и запускается контекст. Параллельно этому запускается 2 таймера для обновления условий и сохранения HyperLogLog. Обязательно в конце указывается awaitTermination(), иначе обработка закончится не начавшись.

JavaRDD<Tuple3<Object, String, Long>> av = HbaseUtil.getEventsHbaseScanRdd(jsc, hbaseConf, new Scan())                 .mapPartitions(it -> {                     ArrayList<Tuple3<Object, String, Long>> list = new ArrayList<>();                     it.forEachRemaining(e -> {                         String pixelId = e.pixelId;                         String vid = e.visitorId;                         long dt = e.date.getTime();                         List<Condition> cond = conditions.get(pixelId);                         if (cond != null) {                             cond.stream()                                     .filter(condition -> e.date.getTime() > beginTime - TimeUnit.DAYS.toMillis(condition.daysInterval)                                             && EventActionUtil.checkEvent(e, condition.condition))                                     .forEach(condition -> list.add(new Tuple3<>(condition.id, vid, dt)));                         }                     });                     return list;                 }).persist(StorageLevel.DISK_ONLY()).setName("RawVisitorAudience"); 

Здесь почти то же, что и в потоковой обработке, но не используется join. Вместо него используется проверка event по списку condition с таким же pixel_id. Как оказалось, такая конструкция требует меньше памяти и выполняется быстрее.

Для начала нужно реализовать метод create() интерфейса JavaStreamingContextFactory и добавить следующие строки при создании контекста

sparkConf.set("spark.streaming.receiver.writeAheadLog.enable", "true"); jssc.checkpoint(checkpointDir); 

Теперь вместо

JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, batchInterval); 

используем

JavaStreamingContext jssc = JavaStreamingContext.getOrCreate(checkpointDir, new ВашРеализованыйКласс()); 

public class UpdateListOutputFormat extends com.aerospike.hadoop.mapreduce.AerospikeOutputFormat<String, Bin> {    private static final Log LOG = LogFactory.getLog(UpdateListOutputFormat.class);     public static class LuaUdfRecordWriter extends AsyncRecordWriter<String, Bin> {         public LuaUdfRecordWriter(Configuration cfg, Progressable progressable) {            super(cfg, progressable);        }         @Override        public void writeAerospike(String key, Bin bin, AsyncClient client, WritePolicy policy, String ns, String sn) throws IOException {            try {                policy.sendKey = true;                Key k = new Key(ns, sn, key);                 Value name = Value.get(bin.name);                Value value = bin.value;                Value[] args = new Value[]{name, value, Value.get(System.currentTimeMillis() / 1000)};                 String packName = AeroUtil.getPackage(cfg);                String funcName = AeroUtil.getFunction(cfg); 		// Execute lua script                client.execute(policy, null, k, packName, funcName, args);            } catch (Exception e) {                LOG.error("Wrong put operation: \n" + e);            }        }    }     @Override    public RecordWriter<String, Bin> getAerospikeRecordWriter(Configuration entries, Progressable progressable) {        return new LuaUdfRecordWriter(entries, progressable);    } } 

Асинхронно выполняет функцию из указанного пакета.

local split = function(str)    local tbl = list()    local start, fin = string.find(str, ",[^,]+$")    list.append(tbl, string.sub(str, 1, start - 1))    list.append(tbl, string.sub(str, start + 1, fin))    return tbl end local save_record = function(rec, name, mp)    local res = list()    for k,v in map.pairs(mp) do        list.append(res, k..","..v)    end     rec[name] = res     if aerospike:exists(rec) then        return aerospike:update(rec)    else        return aerospike:create(rec)    end end function put_in_list_first_ts(rec, name, value, timestamp)     local lst = rec[name]     local mp = map()      if value ~= nil then         if list.size(value) > 0 then             for i in list.iterator(value) do                 mp[i] = timestamp end         end     end      if lst ~= nil then         if list.size(lst) > 0 then             for i in list.iterator(lst) do                 local sp = split(i)                 mp[sp[1]] = sp[2] end         end     end      return save_record(rec, name, mp) end