Introducción
Una empresa minorista recibe registros de pedidos y quiere transformarlos en información útil. Antes de calcular informes de ventas, necesita una forma de representar filas, describir sus columnas y pedir a la herramienta de procesamiento que ejecute trabajo. Comenzará con tres registros lo bastante pequeños como para comprobarlos a simple vista.
Big data, o datos masivos, describe cargas de trabajo cuyo volumen, velocidad de llegada o necesidades de procesamiento pueden superar lo que resulta práctico en una sola máquina. Apache Spark es un motor de procesamiento de datos que puede dividir el trabajo entre máquinas de un clúster. PySpark es su interfaz de Python. Un DataFrame representa filas con columnas con nombre y tipo y admite operaciones como filtrado y agregación. Spark procesa datos; un archivo o sistema de almacenamiento separado proporciona la entrada y la salida persistentes. Este curso comienza con ejemplos diminutos para que pueda razonar sobre la corrección antes de abordar la escala.
Creará una SparkSession, mostrará un DataFrame, examinará su esquema y número de filas y añadirá un registro antes de volver a ejecutar. El entorno preparado utiliza local[2]: Spark se ejecuta en esta única máquina con dos hilos de ejecución. Esto enseña API reales de Spark, mientras que el despliegue distribuido es un tema separado. Necesita conocimientos básicos de variables, listas y llamadas a funciones de Python; la ruta de Python puede ayudarle con esos requisitos previos. No necesita pandas, Hadoop, Hive, Java, instalación ni archivos de laboratorios anteriores. Utilice Terminal.
Ejecute un script completo de un DataFrame pequeño
En este paso, ejecutará un script completo que transforma tres registros de Python en una tabla de Spark. Comience en su espacio de trabajo:
cd /home/labex/project
Una SparkSession es el punto de entrada para crear DataFrames y acceder a las API de datos estructurados de Spark. El constructor establece un nombre de aplicación, selecciona la ejecución local y crea la sesión. local[2] permite dos hilos de ejecución en esta máquina; no crea dos máquinas de clúster. setLogLevel reduce los registros informativos posteriores para que resulte más fácil encontrar la salida de aprendizaje.
En el código Python, from pyspark.sql import SparkSession permite que su programa utilice esa clase instalada. Un punto accede a un miembro con nombre y los paréntesis llaman a un método: df.show() pide a este DataFrame que muestre sus filas. El constructor encadena varias llamadas; los paréntesis exteriores permiten que esa expresión abarque varias líneas. Puede leer cada línea como un ajuste antes de que getOrCreate() obtenga la sesión.
La lista orders contiene tres tuplas, grupos de valores entre paréntesis. Cada tupla proporciona una fila en el mismo orden de las columnas declaradas. La cadena de esquema nombra cada columna y su tipo: identificadores y productos de texto, seguidos de una cantidad entera. createDataFrame utiliza esos registros y ese esquema. show pide a Spark que muestre filas y stop libera la sesión al terminar.
Guarde el script completo. El > del shell escribe el archivo indicado; el marcador PYTHON entre comillas inicia un documento en línea que conserva literalmente el texto Python incluido. Copie hasta la línea PYTHON final:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
El comando preparado bigdata-python selecciona el entorno Python con PySpark instalado. Ejecute el archivo como un programa normal de Python y guarde localmente su salida normal:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
Espere una tabla con las columnas order_id, product y quantity, que contenga s1/notebook/2, s2/pen/1 y s3/folder/3. Pueden seguir apareciendo advertencias de inicio o mensajes de progreso en la terminal; están separados de la tabla guardada. No se garantiza ningún orden de filas sin una ordenación explícita, por lo que debe identificar los registros por sus ID de pedido.
Su programa Python es el controlador (driver): crea la sesión y pide a Spark que realice trabajo. En un clúster, los procesos ejecutores ejecutan tareas sobre porciones de datos. Aquí la ejecución ocurre localmente en la misma máquina. Tiene un programa operativo sin necesidad de configurar un clúster.
Examine el esquema y cuente filas
En este paso, distinguirá lo que puede contener una columna del número de registros de la tabla. Un esquema enumera nombres de columnas, tipos de datos y si admiten valores nulos. Una columna entera representa números enteros adecuados para cálculos de cantidades; el texto tiene otras operaciones. Nullable significa que el campo permite un valor ausente, no que falte en todas las filas.
Añada printSchema para examinar esa descripción y count para calcular el número de filas. Aquí está el archivo actualizado completo para que no exista estado interactivo oculto:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
La tabla debería seguir conteniendo tres registros. Debajo, busque:
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema describe el DataFrame. show y count son acciones que piden a Spark producir un resultado o valor visible. Spark puede aplazar gran parte del procesamiento hasta que una acción solicite una respuesta; laboratorios posteriores explicarán esa ejecución diferida mediante transformaciones y planes. Un conteo de 3 significa tres registros de pedidos, no la suma de cantidades, que respondería a otra pregunta comercial.
Cada invocación inicia un proceso Python nuevo y crea su propia SparkSession. El archivo crea df explícitamente; no depende de que un shell interactivo haya creado spark o df anteriormente.
Prediga y pruebe un registro adicional
En este paso, cambiará los datos conservando su estructura. Prediga qué cambia al añadir s4/notebook/1: el número de filas de pedidos pasa a cuatro, el esquema sigue igual y los tres registros originales permanecen. La fila nueva es un pedido aunque notebook ya aparezca en otro pedido.
Sustituya el script por esta versión, que añade únicamente ese registro:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
Vuelva a ejecutar el archivo en un proceso nuevo:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
Busque los cuatro ID de pedido y Row count: 4. El esquema debería seguir describiendo los mismos tres nombres y tipos de columna. Repetir un nombre de producto no agrupa pedidos ni elimina duplicados automáticamente; esas operaciones deben solicitarse explícitamente en lecciones posteriores.
Los archivos anteriores first-run.txt y schema-and-count.txt permanecen como evidencia de su versión de tres filas. El archivo ejecutable describe ahora la versión de cuatro filas, por lo que otra ejecución independiente puede reproducir ese resultado. Ha completado un pequeño ciclo de cambiar, predecir, ejecutar y comprobar que le ayudará a evaluar flujos de datos mayores, en lugar de confiar únicamente en que un programa haya terminado correctamente.
Resumen
Ejecutó un script real de PySpark con su propia SparkSession, creó un DataFrame con tipos y comprobó sus registros mostrados, esquema y número de filas. Añadir un registro cambió el número de filas conservando el esquema. A continuación, pasará de registros escritos dentro de Python a un archivo CSV con un esquema explícito.
