1

PySpark で Jupyter Notebook を使用しています。その中に、それらの列の列名と型 (整数など) を持つスキーマを持つデータフレームがあります。今、私は flatMap のようなメソッドを使用していますが、これはもはや固定型を持たないタプルのリストを返します。それを達成する方法はありますか？

df.printSchema()
root
 |-- name: string (nullable = true)
 |-- ...
 |-- ...
 |-- ratings: integer (nullable = true)

次に、flatMap を使用して評価値を計算します (ここでは難読化されています)。

df.flatMap(lambda row: (row.id, 5 if (row.ratings > 5) else row.ratings))
y_rate.toDF().printSchema()

そして今、私はエラーが発生します:

TypeError: 次の型のスキーマを推測できません:

スキーマを保持して map/flatMap/reduce を使用する方法はありますか? または、少なくとも特定の型の値を持つタプルを返しますか?

4

1 に答える 1

2

まず、間違った関数を使用しています。flatMapデータが次のようになっているmapとflatten仮定します。

df = sc.parallelize([("foo", 0), ("bar", 10)]).toDF(["id", "ratings"])

の出力はflatMap次と同等になります。

sc.parallelize(['foo', 0, 'bar', 5])

したがって、表示されるエラー。本当に機能させたい場合は、次を使用する必要がありますmap。

df.rdd.map(lambda row: (row.id, 5 if (row.ratings > 5) else row.ratings)).toDF()
## DataFrame[_1: string, _2: bigint]

次に、マッピング オーバーDataFrameは 2.0 ではサポートされなくなりました。rdd最初に抽出する必要があります(df.rdd.map上記を参照)。

最後に、Python と JVM の間でデータを渡すのは非常に非効率的です。対応するシリアライゼーション/デシリアライゼーションとスキーマ推論 (スキーマが明示的に提供されていない場合) を使用して Python と JVM の間でデータを渡す必要があるだけでなく、遅延も解消されます。次のような場合は、SQL 式を使用することをお勧めします。

from pyspark.sql.functions import when

df.select(df.id, when(df.ratings > 5, 5).otherwise(df.ratings))

何らかの理由でプレーンな Python コードが必要な場合は、UDF の方が適している可能性があります。

于 2016-05-14T10:53:36.207 に答える