Pyspark - Funzione di piombo
È possibile partizione delle righe nel frame dati utilizzando la funzione della finestra. È disponibile in pyspark.SQL.finestra modulo.
Sintassi:
dataframe_obj.WithColumn ("Lead_column", lead ("colonna", offset).Over (partizione))Ci vogliono due parametri:
- Il lead_column è il nome della colonna in PysPark DataFrame in cui i valori delle righe con lead sono posizionati in base ai valori in questa colonna.
- L'offset specifica l'intero per restituire quel numero delle righe successive ai valori della riga corrente.
Passaggi:
- Crea un frame dati pyspark che ha alcuni valori simili in almeno una colonna.
- Partizione I dati utilizzando il metodo PartitionBy () disponibili nella funzione della finestra e ordinarli in base alla colonna utilizzando la funzione orderBy ().
Sintassi:
partizione = finestra.Partitionby ("colonna").orderby ("colonna")Possiamo ordinare i dati partizionati con la colonna partizionata o qualsiasi altra colonna.
Ora puoi utilizzare la funzione Lead () sulle righe partizionate usando il Sopra() funzione.
Aggiungiamo una colonna per archiviare il numero di riga usando il withColumn () funzione.
Sintassi:
dataframe_obj.WithColumn ("Lead_column", lead ("colonna", offset).Over (partizione))Qui, il nome specifica il nome della riga e DataFrame_Obj è il nostro PysPark DataFrame.
Implementamoci il codice.
Esempio 1:
Qui, creiamo un frame dati pyspark che ha 5 colonne - ['soggetto_id', 'name', 'age', 'technology1', 'technology2'] con 10 righe e partizione le righe basate sul Tecnologia1 Usando la funzione della finestra. Dopodiché, conduciamo 1 riga.
importare pysparkda pyspark.SQL Import *
Spark_app = SparkSession.costruttore.nome dell'applicazione('_').getOrCreate ()
Studenti = [(4, "Sravan", 23, "Php", "Test"),
(4, "Sravan", 23, "Php", "Test"),
(46, "Mounika", 22, ".Net ',' html '),
(4, "Deepika", 21, "Oracle", "HTML"),
(46, "Mounika", 22, "Oracle", "Test"),
(12, "Chandrika", 22, "Hadoop", "C#"),
(12, "Chandrika", 22, "Oracle", "Test"),
(4, "Sravan", 23, "Oracle", "C#"),
(4, "Deepika", 21, "Php", "C#"),
(46, "Mounika", 22, ".Net ',' test ')
"
DataFrame_obj = Spark_app.CreateAtaFrame (Students, ['soggetto_id', 'name', 'age', 'technology1', 'technology2'])
print ("---------- Frame dati effettivo ----------")
dataframe_obj.spettacolo()
# Importa la funzione della finestra
da pyspark.SQL.finestra di importazione della finestra
#importa il comando da pyspark.SQL.funzioni
da pyspark.SQL.Funzioni Importa lead
#partition Il frame dati in base ai valori nella colonna Technology1 e
#ordina le righe in ogni partizione in base alla colonna soggetto_id
partizione = finestra.Partitionby ("Technology1").orderby ('soggetto_id')
print ("---------- DataFrame partizionato ----------")
#NOW menzionare il lead con offset-1 basato su soggetto_id
dataframe_obj.WithColumn ("Lead", Lead ("soggetto_id", 1).Over (partizione)).spettacolo()
Produzione:
Spiegazione:
Il primo output rappresenta i dati effettivi presenti nel frame dati. Nel secondo output, la partizione viene eseguita in base al Tecnologia1 colonna.
Il numero totale di partizioni è 4.
Partizione 1:
IL .La rete si è verificata due volte nella prima partizione. Dal momento che abbiamo specificato l'offset di lead come 1, l'ultimo .Il valore netto è nullo e il primo .Il valore netto è la prossima riga soggetto_id valore - 46.
Partizione 2:
Hadoop si è verificato una volta nella seconda partizione. Quindi, il piombo è nullo.
Partizione 3:
Oracle si è verificato quattro volte nella terza partizione.
- Per l'ultimo oracolo, il piombo è nullo.
- Per il primo oracolo, il valore del lead è 4 (poiché il valore di successivo riga soggetto_id è 4).
- Per il terzo oracolo, il valore del lead è 12 (poiché il valore di riga di riga successivo è 12).
Per il quarto Oracle, il valore del lead è 46 (poiché il valore di successivo riga soggetto_id è 46).
Partizione 4:
PHP si è verificato tre volte nella quarta partizione.
- Il valore del lead per il 3 ° PHP è nullo.
- Il valore del lead per il 1 ° PHP è 4 (poiché il valore soggetto_id della riga successiva è 4).
- Il valore del lead per il 2 ° PHP è 4 (poiché il valore di successivo riga soggetto_id è 4).
Esempio 2:
Guidare le file di 2. Assicurati di creare il telaio dati PysPark come visualizzato nell'esempio 1.
# Importa la funzione della finestrada pyspark.SQL.finestra di importazione della finestra
#importa il comando da pyspark.SQL.funzioni
da pyspark.SQL.Funzioni Importa lead
#partition Il frame dati in base ai valori nella colonna Technology1 e
#ordina le righe in ogni partizione in base alla colonna soggetto_id
partizione = finestra.Partitionby ("Technology1").orderby ('soggetto_id')
print ("---------- DataFrame partizionato ----------")
#NOW menzionare il lead con offset-2 in base a soggetto_id
dataframe_obj.WithColumn ("Lead", Lead ("soggetto_id", 2).Over (partizione)).spettacolo()
Produzione:
Spiegazione:
La partizione viene eseguita in base al file Tecnologia1 colonna.
Il numero totale di partizioni è 4.
Partizione 1:
IL .La rete si è verificata due volte nella prima partizione. Poiché abbiamo specificato l'offset di lead come 2, l'offset è nullo per entrambi i valori.
Partizione 2:
Hadoop si è verificato una volta nella seconda partizione. Quindi, il piombo è nullo.
Partizione 3:
Oracle si è verificato quattro volte nella terza partizione.
- Negli ultimi due Oracle, il comando è nullo.
- Per il primo oracolo, il valore del lead è 12 (poiché il valore soggetto_id di 2 righe successivi è 12).
- Per il secondo oracolo, il valore del lead è 46 (poiché il valore soggetto_id di 2 righe successivi è 46).
Partizione 4:
PHP si è verificato tre volte nella quarta partizione.
- Negli ultimi due Oracle, il comando è nullo.
- Per il primo PHP, il valore del lead è 4 (poiché il valore soggetto_id di 2 righe successivi è 4).
Esempio 3:
Guidare le righe di 2 in base alla colonna di età. Assicurati di creare il telaio dati PysPark come visualizzato nell'esempio 1.
# Importa la funzione della finestrada pyspark.SQL.finestra di importazione della finestra
#importa il comando da pyspark.SQL.funzioni
da pyspark.SQL.Funzioni Importa lead
#partition Il frame dati in base ai valori nella colonna Technology1 e
#ordina le righe in ogni partizione in base alla colonna soggetto_id
partizione = finestra.Partitionby ("Technology1").orderby ('soggetto_id')
print ("---------- DataFrame partizionato ----------")
#NOW menzionare il lead con offset-2 in base all'età
dataframe_obj.WithColumn ("Lead", Lead ("Age", 2).Over (partizione)).spettacolo()
Produzione:
Spiegazione:
La partizione viene eseguita in base al file Tecnologia1 colonna e piombo sono definiti in base alla colonna di età.
Il numero totale di partizioni è 4.
Partizione 1:
IL .La rete si è verificata due volte nella prima partizione. Poiché abbiamo specificato l'offset di lead come 2, l'offset è nullo per entrambi i valori.
Partizione 2:
Hadoop si è verificato una volta nella seconda partizione. Quindi, il vantaggio è nullo.
Partizione 3:
Oracle si è verificato quattro volte nella terza partizione.
- Per gli ultimi due Oracle, il vantaggio è nullo.
- Per il primo oracolo, il valore del lead è 22 (poiché il valore di età successivo delle 2 file è 22).
- Per il secondo oracolo, il valore del lead è 22 (poiché il valore di età successivo delle 2 righe è 22).
Partizione 4:
PHP si è verificato tre volte nella quarta partizione.
- Per gli ultimi due Oracle, il vantaggio è nullo.
- Per il primo PHP, il valore del lead è 21 (poiché il valore di età successivo delle 2 file è 21).
Conclusione
Abbiamo imparato a ottenere i valori di lead in Pyspark DataFrame nelle righe partizionate. La funzione Lead () in Pyspark è disponibile nel modulo della finestra che viene utilizzato per restituire i valori della riga successiva alle righe di corrente. Prende un parametro di offset che rappresenta il numero totale di righe in modo tale che i valori della riga successiva vengano restituiti alle righe effettive. Per la prima ultima riga/s, vengono posizionati i null (offset). Abbiamo imparato i diversi esempi impostando i diversi offset.