java—为什么ApacheFlink要从数据流中删除事件？

r3i60tvu 于 2021-06-24 发布在 Flink

关注(0)|答案(2)|浏览(378)

在下面的单元测试用例中，由numberofelements指定的一些事件被生成并作为数据流提供。该单元在线路上随机失效。
assertequals（numberofelements，collectsink.values.size（））；
任何解释为什么apache flink跳过这些事件。

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.junit.Before;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

import static java.lang.Thread.sleep;
import static org.junit.Assert.assertEquals;

public class FlinkTest {

StreamExecutionEnvironment env;

@Before
public void setup() {
    env = StreamExecutionEnvironment.createLocalEnvironment();
}

@Test
public void testStream1() throws Exception {
    testStream();
}

@Test
public void testStream2() throws Exception {
    testStream();
}

@Test
public void testStream3() throws Exception {
    testStream();
}

@Test
public void testStream4() throws Exception {
    testStream();
}

@Test
public void testStream() throws Exception {

    final int numberOfElements = 50;

    DataStream<Tuple2<String, Integer>> tupleStream = env.fromCollection(getCollectionOfBucketImps(numberOfElements));
    CollectSink.values.clear();
    tupleStream.addSink(new CollectSink());
    env.execute();
    sleep(2000);

    assertEquals(numberOfElements, getCollectionOfBucketImps(numberOfElements).size());
    assertEquals(numberOfElements, CollectSink.values.size());
}

public static List<Tuple2<String, Integer>> getCollectionOfBucketImps(int numberOfElements) throws InterruptedException {
    List<Tuple2<String, Integer>> records = new ArrayList<>();
    for (int i = 0; i < numberOfElements; i++) {
        records.add(new Tuple2<>(Integer.toString(i % 10), i));
    }
    return records;
}

// create a testing sink
private static class CollectSink implements SinkFunction<Tuple2<String, Integer>> {

    public static final List<Tuple2<String, Integer>> values = new ArrayList<>();

    @Override
    public synchronized void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        values.add(value);
    }
 }
}

例如，teststreamx案例中的任何一个都会随机失败。
上下文：代码以8作为parallelism setu运行，因为它运行的cpu有8个内核

Java apache-flink flink-streaming junit

来源：https://stackoverflow.com/questions/49278579/why-is-apache-flink-droping-the-event-from-datastream

2条答案

按热度按时间

pgpifvop1#

我不知道你工作的平行性（我想这是Flink能分配的最大值）。看起来你可以在Flume的附加值上有一个竞赛条件。
解决方案
我已经运行了您的示例代码，将环境并行度设置为1，一切正常。有关测试的文档示例使用此解决方案链接到文档。

@Before
public void setup() {
    env = StreamExecutionEnvironment.createLocalEnvironment();
    env.setParallelism(1);
}

更好
您可以仅在sink操作符上将并行度设置为1，并保持管道其余部分的并行度。在下面的示例中，我为map操作符添加了一个额外的map函数，强制并行度为8。

public void testStream() throws Exception {

    final int numberOfElements = 50;

    DataStream<Tuple2<String, Integer>> tupleStream = env.fromCollection(getCollectionOfBucketImps(numberOfElements));
    CollectSink.values.clear();
    tupleStream
            .map(new MapFunction<Tuple2<String,Integer>, Tuple2<String,Integer>>() {
                @Override
                public Tuple2<String,Integer> map(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {

                    stringIntegerTuple2.f0 += "- concat something";

                    return stringIntegerTuple2;
                }
            }).setParallelism(8)
            .addSink(new CollectSink()).setParallelism(1);
    env.execute();
    sleep(2000);

    assertEquals(numberOfElements, getCollectionOfBucketImps(numberOfElements).size());
    assertEquals(numberOfElements, CollectSink.values.size());
}

赞(0）回复(0）举报 2021-06-24

7xzttuei2#

当环境的平行性大于1时，存在多个 CollectSink ，这可能导致竞争状况。
以下是避免竞争状况的解决方案：
在类对象上同步

private static class CollectSink implements SinkFunction<Tuple2<String, Integer>> {

    public static final List<Tuple2<String, Integer>> values = new ArrayList<>();

    @Override
    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        synchronized(CollectSink.class) {
            values.add(value);
        }
    }
 }
``` `Collections.synchronizedList()` ```
import java.util.Collections;
private static class CollectSink implements SinkFunction<Tuple2<String, Integer>> {

    public static final List<Tuple2<String, Integer>> values = Collections.synchronizedList(new ArrayList<>());

    @Override
    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        values.add(value);
    }
 }

赞(0）回复(0）举报 2021-06-24

我来回答

java—为什么ApacheFlink要从数据流中删除事件？

2条答案

相关问题

热门标签

最新问答