Appending values to DataSet in Apache Flink

Question

I am currently writing an (simple) analytisis code to sum time connected powerreadings. With the data being assumingly raw (e.g. disturbances from the measuring device have not been calculated out) I have to account for disturbances by calculation the mean of the first one thousand samples. The calculation of the mean itself is not a problem. I only am unsure of how to generate the appropriate DataSet.
For now it looks about like this:

DataSet<Tupel2<long,double>>Gyrotron_1=ECRH.includeFields('11000000000'); // obviously the line to declare the first gyrotron, continues for the next ten lines, assuming separattion of not occupied space
        DataSet<Tupel2<long,double>>Gyrotron_2=ECRH.includeFields('10100000000');
        DataSet<Tupel2<long,double>>Gyrotron_3=ECRH.includeFields('10010000000');
        DataSet<Tupel2<long,double>>Gyrotron_4=ECRH.includeFields('10001000000');
        DataSet<Tupel2<long,double>>Gyrotron_5=ECRH.includeFields('10000100000');
        DataSet<Tupel2<long,double>>Gyrotron_6=ECRH.includeFields('10000010000');
        DataSet<Tupel2<long,double>>Gyrotron_7=ECRH.includeFields('10000001000');
        DataSet<Tupel2<long,double>>Gyrotron_8=ECRH.includeFields('10000000100');
        DataSet<Tupel2<long,double>>Gyrotron_9=ECRH.includeFields('10000000010');
        DataSet<Tupel2<long,double>>Gyrotron_10=ECRH.includeFields('10000000001');
        for (int=1,i<=10;i++) {
            DataSet<double> offset=Gyroton_'+i+'.groupBy(1).first(1000).sum()/1000;
        }

It's the part in the for-loop I'm unsure of. Does anybody know if it is possible to append values to DataSets and if so how?
In case of doubt, I could always put the values into an array but I do not know if that is the wise thing to do.

Arvid Heise Arvid Heise · Accepted Answer · 2020-03-05T19:30:56

This code will not work for many reasons. I'd recommend looking into the fundamentals of Java and the basic data structures and also in Flink.

It's really hard to understand what you actually try to achieve but this is the closest that I came up with

    String[] codes = { "11000000000", ..., "10000000001" };
    DataSet<Tuple2<Long, Double>> result = env.fromElements();
    for (final String code : codes) {
        DataSet<Tuple2<Long, Double>> codeResult = ECRH.includeFields(code)
            .groupBy(1)
            .first(1000)
            .sum(0)
            .map(sum -> new Tuple2<>(sum.f0, sum.f1 / 1000d));
        result = codeResult.union(result);
    }
    result.print();

But please take the time and understand the basics before delving deeper. I also recommend to use an IDE like IntelliJ that would point to at least 6 issues in your code.

Appending values to DataSet in Apache Flink

1 Answers