org. apache .beam.sdk.util.UserCodeException при выполнении Beam Pipeline с использованием Samza Runner - PullRequest
Новая
       95

org. apache .beam.sdk.util.UserCodeException при выполнении Beam Pipeline с использованием Samza Runner

1 голос
/ 11 июля 2020

Я пытаюсь запустить демонстрацию Wordcount из здесь с помощью Samza Runner. Это мой build.gradle 

plugins {
  id 'eclipse'
  id 'java'
  id 'application'

  // 'shadow' allows us to embed all the dependencies into a fat jar.
  id 'com.github.johnrengelman.shadow' version '4.0.3'
}

mainClassName = 'samples.quickstart.WordCount'

maven {
        url = uri('http://packages.confluent.io/maven/')
    }
  mavenCentral()
}

sourceCompatibility = 1.8
targetCompatibility = 1.8

ext.apacheBeamVersion = '2.22.0'

dependencies {
  shadow "org.apache.beam:beam-sdks-java-core:$apacheBeamVersion"
    
  runtime "org.apache.beam:beam-runners-direct-java:$apacheBeamVersion"
  runtime "org.slf4j:slf4j-api:1.+"
  runtime "org.slf4j:slf4j-jdk14:1.+"
  compile group: 'org.apache.beam', name: 'beam-runners-samza', version: '2.22.0'
compile group: 'org.apache.samza', name: 'samza-api', version: '1.4.0'
  compile group: 'org.apache.samza', name: 'samza-core_2.11', version: '1.4.0'
  compile group: 'org.apache.samza', name: 'samza-kafka_2.11', version: '1.4.0'
  compile group: 'org.apache.samza', name: 'samza-kv_2.11', version: '1.4.0'
  compile group: 'org.apache.samza', name: 'samza-kv-rocksdb_2.11', version: '1.4.0'
  testCompile "junit:junit:4.+"
}
shadowJar {
  zip64 true
  baseName = 'WordCount'  // Name of the fat jar file.
  classifier = null       // Set to null, otherwise 'shadow' appends a '-all' to the jar file name.
  manifest {
    attributes('Main-Class': mainClassName)  // Specify where the main class resides.
  }
}

My wordcount. java выглядит следующим образом: 

package samples.quickstart;

import org.apache.beam.runners.samza.SamzaRunner;
//import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.FlatMapElements;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.TypeDescriptors;
import java.util.Arrays;

public class WordCount {
    
    private static final String jobName = "beamtest";
    
  public static void main(String[] args) {
    String inputsDir = "data/*";
    String outputsPrefix = "outputs/part";

    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
    options.setRunner(SamzaRunner.class);

    Pipeline pipeline = Pipeline.create(options);
    
    pipeline
        .apply("Read lines", TextIO.read().from(inputsDir))
        .apply("Find words", FlatMapElements.into(TypeDescriptors.strings())
            .via((String line) -> Arrays.asList(line.split("[^\\p{L}]+"))))
        .apply("Filter empty words", Filter.by((String word) -> !word.isEmpty()))
        .apply("Count words", Count.perElement())
        .apply("Write results", MapElements.into(TypeDescriptors.strings())
            .via((KV<String, Long> wordCount) ->
                  wordCount.getKey() + ": " + wordCount.getValue()))
        .apply(TextIO.write().to(outputsPrefix));
    pipeline.run().waitUntilFinish();
  }
}

Я использую Beam Version 2.22.0. Я пробовал следующие комбинации. Samza 1.4 с Beam 2.22, Samza 1.0 с Beam 2.11 и Beam 2.22 и Samza 0.14.1 с Beam 2.11.0. Однако при выполнении я получаю следующую ошибку: 

java.lang.IncompatibleClassChangeError: Method 'void org.apache.samza.storage.kv.KeyValueStore.deleteAll(java.util.List)' must be InterfaceMethodref constant

Я использую Java 1.8. Есть ли у кого-нибудь идея, что вызывает эту проблему?

1 Ответ

0 голосов
/ 15 июля 2020

вы можете вставить build.gradle и измененный wordcount. java с помощью Samza runner сюда, чтобы мы могли провести расследование, является ли это проблемой несовместимости или проблемой конфигурации. Спасибо, что попробовали Samza runner!

...