В настоящее время я изучаю pyspark и в настоящее время работаю над добавлением столбцов в фреймы данных pyspark, используя несколько условий.
Я пытался работать с пользовательскими функциями, но получаю некоторые ошибки, такие как:
TypeError: 'object' object has no attribute '__getitem__'
from pyspark.sql.functions import udf, struct
from pyspark.sql.types import IntegerType, StringType, FloatType
from pyspark.sql.functions import pandas_udf, PandasUDFType
#first dataframe (superset)
df1 = simple_example1
#second dataframe
df = diff_cols.dropna()
def func(x,y):
z = df1[(df1['a'] == x) & (df1['b'] <= (y+10000000000)) & (df1['b'] >= (y-10000000000))]
z = z[(z["c"] ==1) | (z["d"] ==1)]
z = z[(z["e"]!=0) | (z["f"]!=0) | (z["g"]!=0) | (z["h"]!=0)]
return 1 if z.count() > 3 else 0
udf_func = udf(func, IntegerType())
df = df.withColumn('status', udf_func(df['a'],df['b']))
я пытаюсь сделать следующее:
1. для каждой строки данных фильтра df из df1, где параметр a равен параметру в df, а параметр b должен находиться в диапазоне от b-10 до b + 10
2. затем отфильтруйте эти данные с помощью c или d = 1
3. затем отфильтруйте эти данные, если любой из параметров из e f g h не равен 0
4. затем посчитайте количество строк в подмножестве и назначьте 0/1
5. вернуть это 0/1 в столбце состояния df