PySpark を使用して、既存のデータフレームに新しい列を追加しようとしています。新しい列のエントリは、既存の列に最も近いビン値を表します。以下に示す例では、numpy 配列bucket_arrayはビン (バケット) を表します。
エラーがすぐに言及される PySpark コードの関連セクションを以下に示します。
#Function for finding nearest bucket
def find_nearest(value, bin_array):
bin_array = np.array(list(bin_array))
value = float(value)
idx = np.argmin(np.abs(bin_array - value))
return float(bin_array[idx])
def metric_analyze(entity_peer_labeled_df, metric, delta_weeks, normalize):
# delta_weeks = 1
# normalize = True
# metric : string which denotes column name
# entity_peer_labeled_df : some Pyspark dataframe which has a column titled "pct_difference"
bucket_array = np.arange(-1000, 1000, 5)
udf_nearest_bin = F.udf(find_nearest, T.FloatType())
bucket_df = ( entity_pct_metric_df.withColumn("bucket_array",
F.array(*[F.lit(i) for i in bucket_array])) ).withColumn( "pct_diff_{}_bucket".format(metric) ,
udf_nearest_bin("pct_difference", "bucket_array") )
bucket_df.show()
上記のコードを Jupyter ノートブックで実行すると、正常に動作し、 dataframe を確認できますbucket_df。
それでも、上記のコードを別の python 関数として保存し、Jupyter ノートブックにインポートしてから最後に実行すると、エラーが発生します。行でエラーが発生していることに気付きましたbucket_df.show()。そのエラーの一部を以下に示します。
/mnt1/jupyter/notebooks/username/custom_function.py in metric_analyze(entity_peer_labeled_df, metric, delta_weeks, normalize)
100 udf_nearest_bin("pct_difference", "bucket_array") )
101
--> 102 bucket_df.show()
/usr/lib/spark/python/pyspark/sql/dataframe.py in show(self, n, truncate, vertical)
376 """
377 if isinstance(truncate, bool) and truncate:
--> 378 print(self._jdf.showString(n, 20, vertical))
379 else:
380 print(self._jdf.showString(n, int(truncate), vertical))
/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
完全なエラーはここにあります。
bucket.show()行をに置き換えるとprint( bucket.count() )、エラーは表示されず、正常に実行されます (前述のコードを別の関数として使用しても)。
の例をentity_pct_metric_df以下に示します。
+--------------------+----------+-------------------+-------------------+------------------------------+--------------+
| entity_id|. category| sampled_ts| some_score| some_score_prev_value|pct_difference|
+--------------------+----------+-------------------+-------------------+------------------------------+--------------+
|abccccccccccccccc...| A|2017-12-03 00:00:00| 192| 824| -632.0|
|defffffffffffffff...| A|2017-12-10 00:00:00| 515| 192| 323.0|
|ghiiiiiiiiiiiiiii...| A|2017-12-17 00:00:00| 494| 515| -21.0|
+--------------------+----------+-------------------+-------------------+------------------------------+--------------+
上記のエラーを解決するにはどうすればよいですか?