Co dokładnie robi funkcja .select()?

Question

wpadłem na zaskakującą zachowań podczas korzystania .select():

>>> my_df.show() 
+---+---+---+ 
| a| b| c| 
+---+---+---+ 
| 1| 3| 5| 
| 2| 4| 6| 
+---+---+---+ 

>>> a_c = s_df.select(col("a"), col("c")) # removing column b 
>>> a_c.show() 
+---+---+ 
| a| c| 
+---+---+ 
| 1| 5| 
| 2| 6| 
+---+---+ 

>>> a_c.filter(col("b") == 3).show() # I can still filter on "b"! 
+---+---+ 
| a| c| 
+---+---+ 
| 1| 5| 
+---+---+ 

Takie zachowanie got my zastanawiasz ... Czy moje następujące punkty są prawidłowe?

DataFrames to po prostu widoki, prosta DataFrame to sam widok. W moim przypadku a_c jest po prostu widokiem na my_df.

Po utworzeniu a_c nie utworzono nowych danych, a_c wskazuje tylko te same dane, które wskazuje my_df.

Jeśli istnieją dodatkowe informacje, które są istotne, proszę dodać!

Answer 1

Dzieje się tak ze względu na leniwą naturę Spark. Jest wystarczająco "inteligentny", aby odepchnąć filtr tak, aby zdarzył się na niższym poziomie - przed filtrem *. Tak więc, ponieważ wszystko dzieje się w tym samym stage wykonania i nadal można go rozwiązać. W rzeczywistości można zobaczyć w explain:

== Physical Plan == 
*Project [a#0, c#2] 
+- *Filter (b#1 = 3) <---Filter before Project 
    +- LocalTableScan [A#0, B#1, C#2] 

Można wymusić shuffle i nowego etapu, a następnie zobaczyć Twój filtr nie, choć. Nawet łapanie go podczas kompilacji. Oto przykład:

a_c.groupBy("a","c").count.filter(col("b") === 3) 

* Istnieje również przycinanie projekcja że popycha wybór w dół do warstwy bazy danych, czy zdaje sobie sprawę, że nie potrzebuje kolumny w dowolnym momencie. Wierzę jednak, że filtr sprawiłby, że "potrzebował" go i nie przycinał ... ale tego nie przetestowałem.

Answer 2

Zacznijmy od kilku podstawowych informacji na temat iskry, która jest podstawą. To ułatwi zrozumienie. RDD: Podstawą rdzenia iskrowego jest struktura danych o nazwie RDD, która jest leniwie oceniana jako . Przez leniwą ocenę rozumiemy, że obliczenie RDD dzieje się podczas działania (jak wywoływanie liczby w RDD lub pokazywanie w zbiorze danych).

Zestaw danych lub DataFame (który zestaw danych [wiersz]) również używa RDD w rdzeniu.

Oznacza to, że każda transformacja (jak filtr) zostanie zrealizowana tylko po uruchomieniu akcji (pokaż).

Więc twoje pytanie "Kiedy stworzyłem a_c, żadne nowe dane nie zostały utworzone, a_c wskazuje tylko te same dane, które wskazuje my_df."
Ponieważ nie ma danych, które zostały zrealizowane. Musimy ją zrealizować, aby ją zapamiętać. Twój filtr działa na początkowej ramie danych. Jedynym sposobem, aby twój a_c.filter(col("b") == 3).show() rzucił wyjątek czasu wykonywania jest buforowanie pośredniej ramki danych przy użyciu funkcji ramka danych.cache. Tak więc iskra wyrzuci "główny" org.apache.spark.sql.AnalysisException: Nie można rozpoznać nazwy kolumny Np.

val a_c = s_df.select(col("a"), col("c")).cache 
     a_c.filter(col("b") == 3).show() 

Więc iskra rzuci "main" org.apache.spark.sql.AnalysisException: Można nazwę kolumny nie rozwiązania.