1
私はKafkaからデータを読み込んで処理するSparkアプリケーションを持っています。 mavenコマンドを使用してファットジャーを作成し、コマンドmvn clean compile assembly:singleを作成すると、コマンドツール（No Yarn、スタンドアロンクラスタのみ）を使用してSparkリモートクラスタに正常に送信できます。今では、IntelliJ IDEからファットjarファイルを生成せずに、同じアプリケーションを実行しようとしています。私はIDEでアプリケーションを実行した後には、クラスタのマスターではなく、しばらくのエラーの後にジョブを送信：IDEを使用しているリモートSparkクラスタでSparkアプリケーションを実行する際にエラーが発生する
java.lang.ClassNotFoundException: org.apache.spark.streaming.kafka010.KafkaRDDPartition
私はpom.xmlファイルで依存関係がスパークアプリケーションのためにアクセス可能ではないと思います。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>Saprk</groupId>
<artifactId>SparkPMUProcessing</artifactId>
<version>1.0-SNAPSHOT</version>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<archive>
<manifest>
<mainClass>SparkTest</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.0</version>
</dependency>
</dependencies>
</project>
ポイント：私はリモートクラスタ上のApache FLINKアプリケーションを実行している同じ問題を抱えている。ここ
はpom.xmlファイルです。 Flinkだけでなく、Spark、クラスタに送信するfat jarとterminalコマンドを使用して正しく実行します。
更新：使用方法setJars依存関係のjarファイルと、除外されたjava.lang.ClassNotFoundException:エラータイプを紹介します。今では言う：
java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1.fun$1 of type org.apache.spark.api.java.function.Function in instance of org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1
は、ここに私のコードは次のとおりです。
public class SparkTest {
public static void main(String[] args) throws InterruptedException {
SparkConf conf = new SparkConf().setAppName("PMUStreaming").setMaster("spark://namenode1:7077")
.set("spark.deploy.mode", "client")
.set("spark.executor.memory", "700m").setJars(new String[]{
"/home/xxx/SparkRunningJars/kafka_2.11-0.10.0.0.jar",
"/home/xxx/SparkRunningJars/kafka-clients-0.10.0.0.jar",
"/home/xxx/SparkRunningJars/spark-streaming-kafka-0-10_2.11-2.2.0.jar"
});
Map<String, Object> kafkaParams = new HashMap<>();
Collection<String> TOPIC = Arrays.asList(args[6]);
final String BOOTSTRAPSERVERS = args[0];
final String ZOOKEEPERSERVERS = args[1];
final String ID = args[2];
final int BATCH_SIZE = Integer.parseInt(args[3]);
final String PATH = args[4];
final String READMETHOD = args[5];
kafkaParams.put("bootstrap.servers", BOOTSTRAPSERVERS);
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", ByteArrayDeserializer.class);
kafkaParams.put("group.id", ID);
kafkaParams.put("auto.offset.reset", READMETHOD);
kafkaParams.put("enable.auto.commit", false);
kafkaParams.put("metadata.max.age.ms", 30000);
JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(BATCH_SIZE));
JavaInputDStream<ConsumerRecord<String, byte[]>> stream = KafkaUtils.createDirectStream(
ssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, byte[]>Subscribe(TOPIC, kafkaParams)
);
stream.map(record -> getTime(record.value()) + ":"
+ Long.toString(System.currentTimeMillis()) + ":"
+ Arrays.deepToString(finall(record.value()))
+ ":" + Long.toString(System.currentTimeMillis()))
.map(record -> record + ":"
+ Long.toString(Long.parseLong(record.split(":")[3]) - Long.parseLong(record.split(":")[1])))
.repartition(1)
.foreachRDD(new VoidFunction2<JavaRDD<String>, Time>() {
private static final long serialVersionUID = 1L;
@Override
public void call(JavaRDD<String> rdd, Time time) throws Exception {
if (rdd.count() > 0) {
rdd.saveAsTextFile(PATH + "/" + time.milliseconds());
}
}
});
ssc.start();
ssc.awaitTermination();
}
スパークコードが追加されました。 – soheil
あなたのプロジェクトのjarも 'setJars'に追加する必要があります。私は正しいコードで答えを更新しました。 – Panos
あなたはそうです。私はそれを試みたが、毎回クラスの新しいjarを生成せずにコードを変更できるようにしたい！ – soheil