Блог

У меня есть концепция, которую, я надеюсь, вы сможете помочь прояснить:

В чем разница между следующими тремя способами ссылки на столбец в PySpark dataframe. Я знаю, что в разных ситуациях нужны разные формы, но не уверен, почему.

1. df.col: например F.count(df.col)
2. df[‘col’]: например df['col'] == 0
3. F.col(‘col’): например df.filter(F.col('col').isNull())

Большое спасибо!

В большинстве практических приложений почти нет разницы. Однако они реализуются вызовами разных базовых функций (исходного кода) и, следовательно, не являются точно одинаковыми.

Мы можем проиллюстрировать это небольшим примером:

 df = spark.createDataFrame(
    [(1,'a', 0), (2,'b',None), (None,'c',3)], 
    ['col', '2col', 'third col']
)

df.show()
# ---- ---- --------- 
#| col|2col|third col|
# ---- ---- --------- 
#|   1|   a|        0|
#|   2|   b|     null|
#|null|   c|        3|
# ---- ---- --------- 
  

1. df.col

Это наименее гибкий способ. Вы можете ссылаться только на столбцы, допустимые для доступа с помощью . оператора. Это исключает имена столбцов, содержащие пробелы или специальные символы, и имена столбцов, начинающиеся с целого числа.

Этот синтаксис вызывает df.__getattr__("col") .

 print(df.__getattr__.__doc__)
#Returns the :class:`Column` denoted by ``name``.
#
#        >>> df.select(df.age).collect()
#        [Row(age=2), Row(age=5)]
#
#        .. versionadded:: 1.3
  

Используя . синтаксис, вы можете получить доступ только к первому столбцу этого примера фрейма данных.

 >>> df.2col
  File "<ipython-input-39-8e82c2dd5b7c>", line 1
    df.2col
       ^
SyntaxError: invalid syntax
  

Под капотом он проверяет, содержится ли имя столбца в df.columns , а затем возвращает pyspark.sql.Column указанное.

2. df["col"]

Это вызывает вызов df.__getitem__ . У вас есть некоторая большая гибкость в том смысле, что вы можете делать все, что __getattr__ в ваших силах, плюс вы можете указать любое имя столбца.

 df["2col"]
#Column<2col> 
  

Еще раз, под капотом проверяются некоторые условные обозначения, и в этом случае возвращается pyspark.sql.Column указанное во входной строке.

Кроме того, вы можете передавать несколько столбцов (как list или tuple ) или выражения столбцов.

 from pyspark.sql.functions import expr
df[['col', expr('`third col` IS NULL')]].show()
# ---- ------------------- 
#| col|(third col IS NULL)|
# ---- ------------------- 
#|   1|              false|
#|   2|               true|
#|null|              false|
# ---- ------------------- 
  

Обратите внимание, что в случае нескольких столбцов __getitem__ просто выполняется вызов pyspark.sql.DataFrame.select .

Наконец, вы также можете получить доступ к столбцам по индексу:

 df[2]
#Column<third col>
  

3. pyspark.sql.functions.col

Это собственный способ выбора столбца в Spark, который возвращает expression (это относится ко всем функциям столбца), который выбирает столбец на основе заданного имени. Это полезное сокращение, когда вам нужно указать, что вам нужен столбец, а не строковый литерал.

Например, предположили, что мы хотим создать новый столбец, который принимал бы либо значение из "col" , либо "third col" на основе значения "2col" :

 from pyspark.sql.functions import when

df.withColumn(
    'new', 
    f.when(df['2col'].isin(['a', 'c']), 'third col').otherwise('col')
).show()
# ---- ---- --------- --------- 
#| col|2col|third col|      new|
# ---- ---- --------- --------- 
#|   1|   a|        0|third col|
#|   2|   b|     null|      col|
#|null|   c|        3|third col|
# ---- ---- --------- --------- 
  

Упс, это не то, что я имел в виду. Spark подумал, что мне нужны литеральные строки "col" и "third col" . Вместо этого я должен был написать:

 from pyspark.sql.functions import col
df.withColumn(
    'new', 
    when(df['2col'].isin(['a', 'c']), col('third col')).otherwise(col('col'))
).show()
# ---- ---- --------- --- 
#| col|2col|third col|new|
# ---- ---- --------- --- 
#|   1|   a|        0|  0|
#|   2|   b|     null|  2|
#|null|   c|        3|  3|
# ---- ---- --------- --- 
  

Поскольку функция col() создает выражение столбца без проверки, у этого есть два интересных побочных эффекта.

1. Его можно использовать повторно, поскольку он не зависит от df
2. Его можно использовать до назначения df

 age = col('dob') / 365
if_expr = when(age < 18, 'underage').otherwise('adult')

df1 = df.read.csv(path).withColumn('age_category', if_expr)

df2 = df.read.parquet(path)
    .select('*', age.alias('age'), if_expr.alias('age_category'))
  

age генерирует Column<b'(dob / 365)'>
if_expr генерирует Column<b'CASE WHEN ((dob / 365) < 18) THEN underage ELSE adult END'>