在apache flink中使用DataSet的collect()函数时出现问题

Question

我正在尝试计算社交媒体跟踪图中关系的AdamicAdar指数。我使用apache flink-gelly lirbrarie设置我的边、顶点、数据集和图。下面是我的代码：

    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.DataSource;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.graph.Graph;
    import org.apache.flink.graph.library.similarity.AdamicAdar;
    import org.apache.flink.types.NullValue;
    import org.apache.flink.types.StringValue;
    
    import java.util.List;
    
    public class MyMain {
    
        public static void main(String[] args) {
    
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            DataSource> edgeDataSet = env.
                    readCsvFile(String.valueOf(MyMain.class.getResource("dataset/edges.csv"))).
                    types(StringValue.class, StringValue.class);
            Graph graph = Graph.fromTuple2DataSet(edgeDataSet, env);
    
            List list = null;
            try {
                list = graph.run(new AdamicAdar()).collect();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            System.out.println(list.get(0));
        }
    }

下面是我得到的错误：

    Exception in thread "main" java.lang.NoSuchMethodError: org.apache.flink.configuration.ConfigUtils.decodeListFromConfig(Lorg/apache/flink/configuration/ReadableConfig;Lorg/apache/flink/configuration/ConfigOption;Lorg/apache/flink/util/function/FunctionWithException;)Ljava/util/List;
        at org.apache.flink.client.cli.ExecutionConfigAccessor.getJars(ExecutionConfigAccessor.java:75)
        at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:61)
        at org.apache.flink.client.deployment.executors.LocalExecutor.getJobGraph(LocalExecutor.java:98)
        at org.apache.flink.client.deployment.executors.LocalExecutor.execute(LocalExecutor.java:79)
        at org.apache.flink.api.java.ExecutionEnvironment.executeAsync(ExecutionEnvironment.java:962)
        at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:878)
        at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:862)
        at org.apache.flink.api.java.DataSet.collect(DataSet.java:413)
        at MyMain.main(MyMain.java:23)
    
    Process finished with exit code 1

这也是我使用的edges.csv文件的一部分：

    5 122
    5 156
    5 158
    5 169
    5 180
    5 187
    5 204
    5 213
    5 235
    5 315
    5 316
    6 89
    6 95
    6 147
    6 219
    6 319
    7 22

其中5316表示顶点编号5连接到顶点编号216，这定义了一条边。

这是我的pom.xml文件pom.xml

Answer 1

NoSuchMethodError通常意味着您的依赖项中存在版本不匹配。看看你的pom.xml，你看到的似乎是不同版本的Flink相关库的混合体。

顺便说一句，我还注意到你的.csv没有用逗号分隔。您可能希望简化程序并验证edgeDataset实际上是一个二元组，而不是单个值。