Столбец spark select возвращает ссылку на старый фрейм данных

Question 1

Я использую следующий код:

random = [("ABC",xx, 1), 
          ("DEF",yy,1), 
          ("GHI",zz, 0) 
         ]
randomColumns = ["name","id", "male"]
randomDF = spark.createDataFrame(data=random, schema = randomColumns)
test_df = randomDF.select("name", "id")
test_df.filter(f.col("male") == '1').show()

Из приведенного выше кода я ожидаю, что это приведет к ошибке, потому что для test_df я не выбираю мужской столбец из исходного кадра данных. Удивительно, но приведенный выше запрос выполняется просто отлично без каких-либо ошибок и выводит следующее:

+---------+-------+
|name     |     id|
+---------+-------+
|      abc|     xx|
|      def|     yy|
+---------+-------+

Я хочу понять логику того, что делает искра. В соответствии с документацией spark Select возвращает новый фрейм данных. Тогда почему он все еще может использовать мужской столбец из родительского фрейма данных.

Question 2

Это вызвано DAG, генерируемым Spark. Некоторые операторы (или transformers) выполняются лениво, поэтому они прокладывают путь для Spark для оптимизации DAG.

В этом примере есть два основных шага: select (или project на жаргоне SQL) во-первых, и filter позже. Но на самом деле, при выполнении, filter сначала, а потом select, потому что это более эффективно.

Вы можете проверить этот вывод, сделав explain() метод:

test_df.filter(f.col("flag") == '1').explain()

Он выведет:

== Physical Plan ==
*(1) Project [dept_name#0, dept_id#1L]
+- *(1) Filter (isnotnull(flag#2L) AND (flag#2L = 1))
   +- *(1) Scan ExistingRDD[dept_name#0,dept_id#1L,flag#2L]

Question 3

Добавляя к ответу @chenzhongpu, пожалуйста, обратите внимание, что если вы определяете временное представление поверх своего test_df, запрос завершится ошибкой:

test_df.createOrReplaceTempView("test_df")
spark.sql("select * from test_df where flag = 1").show()
_Traceback (most recent call last): ...
:
pyspark.sql.utils.AnalysisException: u"cannot resolve '`flag`' given input columns: [test_df.dept, test_df.id]; line 1 pos 24;
'Project [*]
 +- 'Filter ('flag = 1)
   +- SubqueryAlias `test_df`
      +- Project [dept#0, id#2L]
         +- LogicalRDD [dept#0, flag#1L, id#2L], false
 _

...потому что а select (=Project узел в плане выполнения) будет предшествовать фильтру (попытка с помощью where пункт).

chenzhongpu · Answer 1 · 2021-11-24T01:29:03

Это вызвано DAG, генерируемым Spark. Некоторые операторы (или transformers) выполняются лениво, поэтому они прокладывают путь для Spark для оптимизации DAG.

В этом примере есть два основных шага: select (или project на жаргоне SQL) во-первых, и filter позже. Но на самом деле, при выполнении, filter сначала, а потом select, потому что это более эффективно.

Вы можете проверить этот вывод, сделав explain() метод:

test_df.filter(f.col("flag") == '1').explain()

Он выведет:

== Physical Plan ==
*(1) Project [dept_name#0, dept_id#1L]
+- *(1) Filter (isnotnull(flag#2L) AND (flag#2L = 1))
   +- *(1) Scan ExistingRDD[dept_name#0,dept_id#1L,flag#2L]

mazaneicha · Answer 2 · 2021-11-24T14:25:52

Добавляя к ответу @chenzhongpu, пожалуйста, обратите внимание, что если вы определяете временное представление поверх своего test_df, запрос завершится ошибкой:

test_df.createOrReplaceTempView("test_df")
spark.sql("select * from test_df where flag = 1").show()
_Traceback (most recent call last): ...
:
pyspark.sql.utils.AnalysisException: u"cannot resolve '`flag`' given input columns: [test_df.dept, test_df.id]; line 1 pos 24;
'Project [*]
 +- 'Filter ('flag = 1)
   +- SubqueryAlias `test_df`
      +- Project [dept#0, id#2L]
         +- LogicalRDD [dept#0, flag#1L, id#2L], false
 _

...потому что а select (=Project узел в плане выполнения) будет предшествовать фильтру (попытка с помощью where пункт).

Столбец spark select возвращает ссылку на старый фрейм данных

Вопрос

Лучший ответ

На других языках

Эта страница на других языках

Популярное в этой категории

Популярные вопросы в этой категории