問題タブ [pyspark]

一見単純な質問ですが、答えが見つかりません。

問題: 1 つのフィールドを取り、そこから 3 つのフィールドを作成する map() に渡す関数を作成します。map() の出力で、入力 RDD と新しい/出力 RDD の両方のフィールドを含む新しい RDD を取得したいと考えています。どうすればいいですか？

より多くの出力 RDD を元の RDD に結合できるように、データのキーを関数の出力に追加する必要がありますか? それは適切な/ベストプラクティスですか？

さらに基本的なことですが、2 つの Row を組み合わせることができないようです。

これは、私が望むように new Row() を返しません。

ありがとう

次のデータセットがあるとしましょう

私がやろうとしているのは、値を列インデックスであるキーと値である値にフラットマップすることです。誰でも私にガイダンスを与えることができますか？列のインデックスを取得するのは難しいと思います。

非常に単純なテキスト解析と関数型プログラミング機能のために python を活用したいと思います。また、numpy や scipy などの科学計算ライブラリの豊富な機能を活用したいと考えているため、タスクに pyspark を使用したいと考えています。

最初に実行しようとしているタスクは、ストリームの一部として書き込まれているテキスト ファイルがあるバケットから読み取ることです。誰かが pyspark を使用して s3 パスからストリーミング データを読み取る方法のコード スニペットを貼り付けることができますか? これは最近まで scala と java を使用してのみ実行できると思っていましたが、今日、spark 1.2 以降、ストリーミングが pyspark でもサポートされていることを知りましたが、S3 ストリーミングがサポートされているかどうかは不明です?

私がscalaで行っていた方法は、HadoopTextFileとして読み込んで、構成パラメーターを使用してawsキーとシークレットを設定することです。pysparkで同様のことを行うにはどうすればよいですか?

どんな助けでも大歓迎です。

前もって感謝します。

2つid: bigintあり、1つを削除したいです。どのようにできるのか？

1 つはランダム ログを生成し、kafka プロデューサーにフィードし、もう 1 つは kafka トピックを使用して Spark Streaming でデータのストリームを作成します。

残念ながら、スクリプトを実行すると、KafkaUtils.createStream に関連するエラーが発生します...

これが私の端末の表示です：

私の2つの機能：

ご協力いただきありがとうございます ！

テスト機械学習データセットを使用して、pyspark で Spark MLlib パッケージを実行しようとしています。データセットを半分のトレーニングデータセットと半分のテストデータセットに分割しています。以下は、モデルを構築する私のコードです。ただし、すべての従属変数にわたる NaN、NaN.. の重みを示しています。理由がわかりませんでした。しかし、StandardScaler 関数を使用してデータを標準化しようとすると機能します。

助けてくれてありがとう。

以下は、スケーリングを行うために使用したコードです。