問題タブ [py4j]

py4jを使用してpysparkでJava関数を実行しようとしていました。Py4j を使用すると、JVM で Java オブジェクトにアクセスできます。JVM の別のインスタンスを作成し、Java 関数を正常に実行できました。

py4j は、GatewayServer インスタンスを介してこの通信を有効にします。

どうにかして Spark の内部 JVM にアクセスして Java 関数を実行できるかどうか疑問に思っていました。spark の py4j Gatewayserver のエントリ ポイントは何ですか? 関数をエントリ ポイントに追加するにはどうすればよいですか?

H2Oのスパークリングウォーターを使ってモデルを作成しました。そして今、それを巨大なSpark DF（スパースベクトルが取り込まれた）に適用したいと思います。私はpythonとpyspark、pysparklingを使用しています。基本的に、内部で model.predict() 関数を使用してマップ ジョブを実行する必要があります。しかし、データを H2O コンテキストにコピーすることは大きなオーバーヘッドであり、オプションではありません。私がやろうと思っているのは、h2o モデルから POJO (Java クラス) モデルを抽出し、それを使用してデータフレームでマップを行うことです。私の質問は次のとおりです。

1. より良い方法はありますか？
2. Java クラスの pyspark ラッパーの作成方法。そこからメソッド .score(double[] data, double[] result) を 1 つだけ使用する予定です。
3. Spark ML ライブラリのラッパーを最大限に再利用するには?

ありがとうございました！

内部計算に Java オブジェクトを使用する pyspark python で使用する UDF を作成する必要があります。

それが単純なpythonであれば、私は次のようにします：

次を使用して呼び出します。

ただし、必要な関数の実装は Java であり、Python ではありません。Pythonから同様の方法で呼び出すことができるように、何らかの形でラップする必要があります。

私の最初の試みは、Javaオブジェクトを実装してから、それをpysparkのpythonでラップし、それをUDFに変換することでした。シリアル化エラーで失敗しました。

Java コード:

pyspark コード:

エラー：

EDIT : Java クラスをシリアライズ可能にしようとしましたが、役に立ちませんでした。

私の2番目の試みは、最初にJavaでUDFを定義することでしたが、正しくラップする方法がわからないため失敗しました:

Java コード: パッケージ com.test1.test2;

しかし、どのように使用しますか？私は試した：

しかし、私は得る:

そして、次の代わりに a.call を使用しようとしました:

しかし得た: ----------------------------------------------- ---------------------------- TypeError トレースバック (最新の呼び出しが最後) in () 3 a = sc._gateway.jvm.com. test1.test2.TestClassUdf(7) 4 dfint = sqlContext.range(0,15) ----> 5 df = dfint.withColumn("a",a.call(dfint.id))

どんな助けでも感謝します。

私は py4j を使用して、ユーザーフレンドリーではない Java ライブラリーを中心にユーザーフレンドリーな Python ライブラリーを構築してきました。ほとんどの場合、これは簡単で、py4j は優れたツールです。ただし、Python と Java の間で行列を送信するときに問題が発生しました。

具体的には、整数行列を引数として受け入れるJavaの静的関数があります。

次のように Py4j からこれを呼び出せるようにしたいと思います。

これは機能しませんが、それほど驚くことではありません。また、numpy_matrix代わりにプレーンな python リストのリストに変換された場合も機能しません。関数呼び出しの前に、Java 配列を作成してデータを転送することで解決できると思っていました。

これで、このコードは正しく実行されます。ただし、実行には約2 分かかります。ちなみに、私が扱っている行列は (3 x ~300,000) 要素のオーダーです。

行列を変換するだけで信じられないほどの時間を必要としない、Py4jでこれを行う標準的な方法はありますか? 1、2 秒かかることは気にしませんが、これでは遅すぎます。Py4j がこの種の通信用にセットアップされていない場合、Python 用の Java 相互運用ライブラリはありますか?

注: Java ライブラリは、int[][]行列を不変の配列として扱います。つまり、変更しようとはしません。

Scala に 2 次元のリストがあり、dataListそれを Pandas に変換したいと考えていDataFrameます。

を印刷したい場合dataList、すべてが正常に機能し、Python のオブジェクトのタイプは次のとおりです。

私の知る限りpy4j、Python でコレクションにアクセスするには を使用する必要があります。ただし、ネストされたループを使用して の各要素にアクセスdataListし、DataFrame. それを直接DataFrameまたはpythonリストに変換してから、よりスマートな方法はありますDataFrameか？

Zeppelin 0.5.5 を使用しています

ログが INFO レベルで開始されると、ログに大量のpy4j.java_gateway:Received command c on object id p0ログが記録されます。どうすれば非表示にできますか?

Spark Streaming は初めてです。PyCharm で PySpark を使用すると、socketTextStream の初期化を渡すことができません。

次のエラーが表示されます。問題が何であるかについてのアイデアはありますか？

トレースバック (最後の最後の呼び出し): ファイル "/Users/egorkorneev/PycharmProjects/BasicStreaming/SparkConsumer.py"、75 行目、dc.start_streaming() ファイル "/Users/egorkorneev/PycharmProjects/BasicStreaming/SparkConsumer.py"、行40, start_streaming の項目 = stream.socketTextStream('localhost', '9009') ファイル "/Users/egorkorneev/rnd/spark-1.6.1-bin-hadoop2.6/python/pyspark/streaming/context.py",行 352、socketTextStream return DStream(self._jssc.socketTextStream(hostname, port, jlevel), self, File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/site-packages/py4j/ java_gateway.py"、933 行目、呼び出し中answer, self.gateway_client, self.target_id, self.name) ファイル "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/site-packages/py4j/protocol.py"、316 行目get_return_value format(target_id, ".", name, value)) py4j.protocol.Py4JError: o19.socketTextStream の呼び出し中にエラーが発生しました。トレース: py4j.Py4JException: メソッド socketTextStream([class java.lang.String, class java.lang.String, class org.apache.spark.storage.StorageLevel]) は py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine. java:335) で py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:344) で py4j.Gateway.invoke(Gateway.java:252) で py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133) で py4j. commands.CallCommand.execute(CallCommand.java:79) を py4j.GatewayConnection で実行します。

スパーク 1.6.1