Condividi tramite

Inviare e gestire processi in un cluster Apache Spark™ in HDInsight nel servizio Azure Kubernetes

  • Articolo

Nota

Azure HDInsight su AKS verrà ritirato il 31 gennaio 2025. Prima del 31 gennaio 2025, sarà necessario eseguire la migrazione dei carichi di lavoro a Microsoft Fabric o a un prodotto Azure equivalente per evitare interruzioni improvvise dei carichi di lavoro. I cluster rimanenti nella sottoscrizione verranno arrestati e rimossi dall’host.

Solo il supporto di base sarà disponibile fino alla data di ritiro.

Importante

Questa funzionalità è attualmente disponibile solo in anteprima. Le Condizioni per l'utilizzo supplementari per le anteprime di Microsoft Azure includono termini legali aggiuntivi che si applicano a funzionalità di Azure in versione beta, in anteprima o in altro modo non ancora disponibili a livello generale. Per informazioni su questa anteprima specifica, vedere Informazioni sull'anteprima di Azure HDInsight nel servizio Azure Kubernetes. Per domande o suggerimenti sulle funzionalità, inviare una richiesta in AskHDInsight con i dettagli e seguire Microsoft per altri aggiornamenti nella Community di Azure HDInsight.

Dopo aver creato il cluster, l'utente può usare varie interfacce per inviare e gestire i processi tramite

  • uso di Jupyter
  • uso di Zeppelin
  • uso di ssh (spark-submit)

Uso di Jupyter

Prerequisiti

Un cluster Apache Spark™ in HDInsight nel servizio Azure Kubernetes. Per altre informazioni, vedere Creare un cluster Apache Spark.

Jupyter Notebook è un ambiente notebook interattivo che supporta diversi linguaggi di programmazione.

Creare un notebook di Jupyter Notebook

  1. Passare alla pagina del cluster Apache Spark™ e aprire la scheda Panoramica. Fare clic su Jupyter e chiede di eseguire l'autenticazione e aprire la pagina Web di Jupyter.

    Screenshot di come selezionare Jupyter Notebook.

  2. Nella pagina Web di Jupyter selezionare Nuovo > PySpark per creare un notebook.

    Screenshot della nuova pagina PySpark.

    Nuovo notebook creato e aperto con il nome Untitled(Untitled.ipynb).

    Nota

    Usando il kernel PySpark o Python 3 per creare un notebook, la sessione Spark viene creata automaticamente quando si esegue la prima cella di codice. Non è necessario creare in modo esplicito la sessione.

  3. Incollare il codice seguente in una cella vuota di Jupyter Notebook e quindi premere MAIUSC+INVIO per eseguire il codice. Vedere qui per altri controlli su Jupyter.

    Screenshot della pagina PySpark con il contenuto. 

    %matplotlib inline
import pandas as pd
import matplotlib.pyplot as plt
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row
s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   
#create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   
#create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])

  4. Tracciare un grafico con stipendio e età come assi X e Y

  5. Incollare nello stesso notebook il codice seguente in una cella vuota di Jupyter Notebook e quindi premere MAIUSC+INVIO per eseguire il codice.

    %matplotlib inline
import pandas as pd
import matplotlib.pyplot as plt

plt.plot(age_series,salary_series)
plt.show()

    Screenshot dell'output del grafico.

Salvare il notebook

  1. Dalla barra dei menu del notebook passare a File > Salva e checkpoint.

  2. Arrestare il notebook per rilasciare le risorse cluster: dalla barra dei menu del notebook, passare a File >Chiudi e interrompi. È anche possibile eseguire uno qualsiasi dei notebook nella cartella degli esempi.

    Screenshot di come salvare i libri delle note.

Uso di notebook di Apache Zeppelin

I cluster Apache Spark in HDInsight nel servizio Azure Kubernetes includono Notebook Apache Zeppelin. Usare i notebook per eseguire processi Apache Spark. Questo articolo illustra come usare il notebook Zeppelin in un cluster HDInsight nel servizio Azure Kubernetes.

Prerequisiti

Un cluster Apache Spark in HDInsight nel servizio Azure Kubernetes. Per istruzioni, vedere Creare un cluster Apache Spark.

Avviare un notebook di Apache Zeppelin

  1. Passare alla pagina Panoramica del cluster Apache Spark e selezionare Notebook Zeppelin nei dashboard del cluster. Richiede di eseguire l'autenticazione e aprire la pagina Zeppelin.

    Screenshot di come selezionare Zeppelin.

  2. Creare un nuovo notebook. Nel riquadro intestazione passare a Notebook > Crea nuova nota. Verificare che l'intestazione del notebook mostri uno stato connesso. Indica un punto verde nell'angolo superiore destro.

    Screenshot di come creare un notebook zeppelin.

  3. Eseguire il codice seguente in Zeppelin Notebook:

    %livy.pyspark
import pandas as pd
import matplotlib.pyplot as plt
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row

s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   #create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   #create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])

  4. Selezionare il pulsante Riproduci per il paragrafo per eseguire il frammento di codice. Lo stato nell'angolo destro del paragrafo deve passare da PRONTO, IN ATTESA, IN ESECUZIONE, a COMPLETATO. L'output viene visualizzato nella parte inferiore dello stesso paragrafo. Lo screenshot è simile all'immagine seguente:

    Screenshot di come eseguire il notebook Zeppelin.

    Output:

    Screenshot dell'output del notebook Zeppelin.

Uso dei processi di invio di Spark

  1. Creare un file usando il seguente comando `#vim samplefile.py'

  2. Questo comando apre il file vim

  3. Incollare il codice seguente nel file vim

    import pandas as pd
import matplotlib.pyplot as plt

From pyspark.sql import SparkSession
Spark = SparkSession.builder.master('yarn').appName('SparkSampleCode').getOrCreate()
# Initialize spark context
data1 = [22,40,10,50,70]
s1 = pd.Series(data1)   #One-dimensional ndarray with axis labels (including time series).

data2 = data1
index = ['John','sam','anna','smith','ben']
s2 = pd.Series(data2,index=index)

data3 = {'John':22, 'sam':40, 'anna':10,'smith':50,'ben':70}
 s3 = pd.Series(data3)

s3['jp'] = 32     #insert a new row

s3['John'] = 88

names = ['John','sam','anna','smith','ben']
ages = [10,40,50,48,70]
name_series = pd.Series(names)
age_series = pd.Series(ages)

data_dict = {'name':name_series, 'age':age_series}
dframe = pd.DataFrame(data_dict)   #create a pandas DataFrame from dictionary

dframe['age_plus_five'] = dframe['age'] + 5   #create a new column
dframe.pop('age_plus_five')
#dframe.pop('age')

salary = [1000,6000,4000,8000,10000]
salary_series = pd.Series(salary)
new_data_dict = {'name':name_series, 'age':age_series,'salary':salary_series}
new_dframe = pd.DataFrame(new_data_dict)
new_dframe['average_salary'] = new_dframe['age']*90

new_dframe.index = new_dframe['name']
print(new_dframe.loc['sam'])

  4. Salvare il file con il metodo seguente.

    1. Premere il pulsante Escape
    2. Immettere il comando :wq

  5. Eseguire il comando seguente per eseguire il processo.

    /spark-submit --master yarn --deploy-mode cluster <filepath>/samplefile.py

    Screenshot che mostra come eseguire il processo di invio spark.

Monitorare le query in un cluster Apache Spark in HDInsight nel servizio Azure Kubernetes

Interfaccia utente della cronologia Spark

  1. Fare clic sull'interfaccia utente del server cronologia Spark nella scheda Panoramica.

    Screenshot che mostra l'interfaccia utente di Spark.

  2. Selezionare l'esecuzione recente dall'interfaccia utente usando lo stesso ID applicazione.

    Screenshot che mostra come eseguire l'interfaccia utente di Spark.

  3. Visualizzare i cicli del grafico aciclico diretto e le fasi del processo nell'interfaccia utente del server cronologia Spark.

    Screenshot del ciclo DAG.

Interfaccia utente della sessione Livy

  1. Per aprire l'interfaccia utente della sessione Livy, digitare il comando seguente nel browser https://<CLUSTERNAME>.<CLUSTERPOOLNAME>.<REGION>.projecthilo.net/p/livy/ui

    Screenshot di come aprire l'interfaccia utente della sessione livy.

  2. Visualizzare i log del driver facendo clic sull'opzione driver nei log.

Interfaccia utente Yarn

  1. Nella scheda Panoramica fare clic su Yarn e aprire l'interfaccia utente di Yarn.

    Screenshot di come selezionare l'interfaccia utente di Yarn.

  2. È possibile tenere traccia del processo eseguito di recente dallo stesso ID applicazione.

  3. Fare clic sull'ID applicazione in Yarn per visualizzare i log dettagliati del processo.

    Visualizzare i log.

Riferimento