Как отфильтровать значение, превышающее определенную точку во флинке? - PullRequest
Новая
       8

Как отфильтровать значение, превышающее определенную точку во флинке?

0 голосов
/ 12 февраля 2019

У меня есть два потока.Первый - это поток на основе времени, и я использовал countTimeWindow, чтобы получить первые 10 точек данных для расчета значения статистики.Я вручную использовал переменную cnt, чтобы сохранить только первое окно, и отфильтровал оставшиеся значения, как показано в приведенном ниже коде.

И затем я хочу использовать это значение для фильтрации основного потока, чтобызначения, которые превышают значение stat, которое я вычислил в потоке окна.

Однако я не знаю, как объединить или рассчитать эти два потока для достижения моей цели.

Мой сценарий таков: если я преобразую первое значение статистики в переменную широковещания, я передаю его основному потоку, чтобы иметь возможность фильтровать входящие значения на основе значения статистики в широковещании.переменная.

Ниже мой код. 

import com.sun.org.apache.xpath.internal.operations.Bool;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.windowing.*;
import org.apache.flink.util.Collector;
import scala.Int;


import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

public class ReadFromKafka {
    static int cnt = 0;
    public static void main(String[] args) throws Exception{
        // create execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink");

        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer09<>("flinkStreaming11", new SimpleStringSchema(), properties));

        env.enableCheckpointing(1000);

//Time based window stream
        DataStream<String> process = stream.countWindowAll(10).process(new ProcessAllWindowFunction<String, Tuple2<Double, Integer>, GlobalWindow>() {
            @Override
            public void process(Context context, Iterable<String> iterable, Collector<Tuple2<Double, Integer>> collector) throws Exception {
                Double sum = 0.0;

                int n = 0;
                List<Double> listDouble = new ArrayList<>();
                for (String in : iterable) {
                    n++;
                    double d = Double.parseDouble(in);
                    sum += d;
                    listDouble.add(d);
                }

                cnt++;
                Double[] sd = listDouble.toArray(new Double[listDouble.size()]);
                double mean = sum / n;

                double sdev = 0;
                for (int i = 0; i < sd.length; ++i) {
                    sdev += ((sd[i] - mean) * (sd[i] - mean)) / (sd.length - 1);
                }
                double standardDeviation = Math.sqrt(sdev);
                collector.collect(new Tuple2<Double, Integer>(mean + 3 * standardDeviation, cnt));
            }
        }).filter(new FilterFunction<Tuple2<Double, Integer>>() {
            @Override
            public boolean filter(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                Integer i1 = doubleIntegerTuple2.f1;
                if (i1 > 1)
                    return false;
                else
                    return true;
            }
        }).map(new RichMapFunction<Tuple2<Double, Integer>, String>() {
            @Override
            public String map(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                return String.valueOf(doubleIntegerTuple2.f0);
            }
        });



//I don't think that this is not a proper solution.
        process.union(stream).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                return false;
            }
        })

        env.execute("InfluxDB Sink Example");

        env.execute();
    }
}

1 Ответ

0 голосов
/ 12 февраля 2019

Во-первых, я думаю, что у вас есть только один поток, верно?Есть только один источник удвоений на основе Kafka (закодированный как Strings).

Во-вторых, если первые 10 значений действительно постоянно определяют предел для фильтрации, то вы можете просто запустить поток в функцию RichFlatMap, гдевы фиксируете первые 10 значений для расчета вашего максимального значения, а затем фильтруете все последующие значения (только выходные значения> = этот предел).

Обратите внимание, что обычно вы хотите сохранить состояние (массив из 10 начальных значений)(плюс предел), чтобы ваш рабочий процесс можно было перезапустить с контрольной точки / точки сохранения.

Если вместо этого вы постоянно пересчитываете свой предел из последних 10 значений, то код становится немного более сложнымв том, что у вас есть очередь значений, и вам нужно выполнить фильтрацию по значению, удаляемому из очереди, когда вы добавляете новое значение.

...