問題タブ [kryo]

kryo を使用して、それ自体がいくつかのカスタム オブジェクト、より具体的にはカスタム オブジェクトの HashMap を含むカスタム クラスをシリアル化しようとしています。このようなものを処理する適切な方法を考えていました。以下は、シリアル化しようとしているクラス (Data)、ネストされたクラス、および現在の kryo 実装です。これは正しいアプローチですか？

次のスニペットに示すように、RDD の各パーティションがそのオブジェクトにアクセスできるように、ドライバー ノードから RDD が存在する他のノードにオブジェクトを渡したいと考えています。

上記のスニペットでは、Spark で Kryo によってImmutableBytesWritableをシリアル化しようとしたため、次のようにしました。

1. spark コンテキストに渡されるSparkConfインスタンスを構成します。つまり、「 spark.serializer」を「org.apache.spark.serializer.KryoSerializer」に設定し、「spark.kryo.registrator」を「xt.HelloKryoRegistrator 」に設定します。
2. ImmutableBytesWritableクラスを登録するカスタム Kryo 登録クラスを作成します。
3. ImmutableBytesWritableのシリアライザーを作成する

ただし、Spark アプリケーションを yarn-client モードで送信すると、次の例外がスローされました。

スレッド "メイン" org.apache.spark.SparkException での例外: タスクは org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166) でシリアライズできません org.apache.spark.util.ClosureCleaner$.clean で(ClosureCleaner.scala:158) org.apache.spark.SparkContext.clean(SparkContext.scala:1242) で org.apache.spark.rdd.RDD.map(RDD.scala:27​​0) で xt.HelloSpark$.main で(HelloSpark.scala:23) xt.HelloSpark.main(HelloSpark.scala) で sun.reflect.NativeMethodAccessorImpl.invoke0(ネイティブ メソッド) で sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) で sun.reflect. org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:325) org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75) で org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) で 原因: java.io.NotSerializableException: org.apache.hadoop.hbase.io.ImmutableBytesWritable at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) at java.io.ObjectOutputStream.writeSerialData( ObjectOutputStream.java:1508) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) で java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) で java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) で) org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) で org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73) at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164) ... 12 詳細

ImmutableBytesWritableは Kryo でシリアル化できないようです。では、Kryo を使用して Spark にオブジェクトをシリアル化させる正しい方法は何でしょうか? Kryo は任意の型をシリアル化できますか?

サーバーからクライアントに送信しているグラフのようなオブジェクトがあり、これにnodesはadjacencyLists.

私はこれに似たものを持っています:

これらの消去を含むオブジェクトを送信するたびに、次のエラーが表示されます。

Kryonet でこれに対する回避策はありますか? ありがとう

Android で Kryonet を使用していますが、起動後に VM が次のエラーでクラッシュします。

何が問題ですか？

SchemaRDD のカスタム タイプを導入したいのですが、この例に従っ
ていますが、Kryo シリアル化の問題が発生しています。スタック トレースは次のとおりです。

com.gis.io.GeometryWritableは、APP jar に存在する UDT を作成するためのカスタム クラスです。ここで Google グループで使用されなくなったように、spark のデフォルト シリアライザーを kryo から Java に変更してみました。しかし、それは私を助けません。助言がありますか ？？何か足りない場合は？以下は私のクラスです：

これが私がそれを使用している方法です。

編集: 実際、RDD (テーブル) を 1 つだけ使用すると、カスタム UDT が正常に機能することがわかりました。この問題は、2 つ以上の RDD に参加するときに発生します。thisによると、カスタム ROW を使用して JOIN を使用する場合のバグです。手がかりはありますか??

これは、Kryo シリアライゼーション ライブラリを使用しています。

で DB にシリアル化されたオブジェクトを既に持っていますFieldSerializer。ここで、シリアル化されたクラスにいくつかのフィールドを追加し、それらとの下位互換性を確保したいと考えています。

これは可能ですか？それを行う最良の方法は何ですか？

デシリアライザーを に切り替えてみましたTaggedFieldSerializerが、当然のことながら、それは機能しません ( Encountered unregistered class ID:)。

追加のフィールドを持つ同じクラスを使用して、これらのオブジェクトを逆シリアル化する方法はありますか?

更新 1

明らかでない場合に備えて、以前にシリアル化されたすべてのオブジェクトを DB から削除しないようにしたいと考えています。

kryo を使用した特定の avro シリアライゼーションについて質問があります。この質問が冗長に思われる場合は申し訳ありませんが、簡単な答えが見つかりませんでした。

Kryo で実行するように Spark を構成しましたが、うまく動作します。Avro で生成されたクラスは、カスタムの KryoRegistrator クラスを使用して適切にシリアル化されます。私もtwitter.chill-avroパッケージを使っています。

それにもかかわらず、 a をシリアライズしようとするとMap[String, MyAvroClass]、シリアライゼーションを機能させる唯一の方法は、を使用して拡張MyAvroClassすることjava.io.Serializableです。

誰かが実際に何が起こっているのかを説明し、これ Map[..., AvroClass]を kryo に登録する方法を教えてもらえますか?

ありがとうございました

シリアル化の有無にかかわらず実行時間を測定し、kryo オブジェクト ストリームをディスクに保存して、spark でキャッシュをシミュレートするために、Apache Spark で Kryo シリアル化をテストしようとしています。

私が設計したテストは、シリアル化された CassandraRDD オブジェクトに Cassandra テーブルを格納することです。

CassandraRDD を生成する Scala コードは次のとおりです。

このコードは機能しますが、CassandraRDD オブジェクトである kvRDD がシリアル化されていないと思われます。

Kryoでシリアル化できるものとできないものに関する規則はありますか? このクラスを kryo.register に登録するにはどうすればよいですか?

kryo.register(ClassOf[CassandraRDD]) で登録しようとすると、実行しようとすると次のエラーが発生します。

私はScalaとKryoに非常に慣れていないことに注意してください。

よろしくお願いします