Faísca selecione a coluna retorna a referência do antigo dataframe

Question 1

Eu uso o seguinte código:

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()

A partir do código acima eu esperava que resultar em um erro porque, para o test_df eu não selecione o macho coluna do original dataframe. Surpreendentemente, a consulta acima funciona muito bem, sem qualquer erro e emite o seguinte:

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

Eu quero entender a lógica por trás do que de ignição está fazendo. Como por faísca documentação Select retorna um novo dataframe. Então por que ainda é capaz de usar o masculino coluna do pai dataframe.

Question 2

Isso é causado pelo DAG gerado por Centelha. Alguns operadores (ou transformers) é são preguiçosos-executado, para que eles pavimentar o caminho para a Faísca para otimizar o DAG.

Neste exemplo, há duas etapas principais: select (ou project no SQL do jargão） primeiro, e filter mais tarde. Mas na verdade, quando da execução, filter primeiro e, em seguida, selectporque é muito mais eficiente.

Você pode verificar essa conclusão por explain() método:

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

O resultado será:

== 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

Adicionar a @chenzhongpu 's resposta, por favor, note que se você definir um temp vista em cima do seu test_df, a consulta irá falhar:

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
 _

...porque um select (=Project nó no plano de execução) vai preceder filtro (tentativa de via where cláusula).

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

Isso é causado pelo DAG gerado por Centelha. Alguns operadores (ou transformers) é são preguiçosos-executado, para que eles pavimentar o caminho para a Faísca para otimizar o DAG.

Neste exemplo, há duas etapas principais: select (ou project no SQL do jargão） primeiro, e filter mais tarde. Mas na verdade, quando da execução, filter primeiro e, em seguida, selectporque é muito mais eficiente.

Você pode verificar essa conclusão por explain() método:

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

O resultado será:

== 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

Adicionar a @chenzhongpu 's resposta, por favor, note que se você definir um temp vista em cima do seu test_df, a consulta irá falhar:

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
 _

...porque um select (=Project nó no plano de execução) vai preceder filtro (tentativa de via where cláusula).

Faísca selecione a coluna retorna a referência do antigo dataframe

Pergunta

Melhor resposta

Em outros idiomas

Esta página está em outros idiomas

Popular nesta categoria

Perguntas populares nesta categoria