Uso del conector HDFS con Spark

Introducción

Este artículo proporciona un tutorial que ilustra el uso del conector del sistema de archivos distribuido de Hadoop (HDFS) con el marco de aplicación Spark. Para el tutorial, usamos el sistema operativo Oracle Linux 7.4 y ejecutamos Spark como un sistema autónomo en una sola computadora.

Requisitos previos

A continuación, se enumeran los requisitos para completar el tutorial:

Debe tener permiso para crear una instancia informática. Para obtener instrucciones, consulte Creación de una instancia.
Debe poder conectarse a la instancia de servicio que ha iniciado. Para obtener instrucciones, consulte Conexión a una instancia.
Debe disponer del OCID, la huella y el clave privada adecuados para el usuario IAM (Identity and Access Management) que utilizará para interactuar con Object Storage. Para obtener instrucciones, consulte Configuración y requisitos.
Debe tener un cubo de Object Storage con el que pueda conectarse.
El usuario de IAM debe poder leer y escribir en ese cubo mediante la consola.

Uso de Spark

Instalación de Spark y dependencias

Nota

Para realizar este ejemplo, instale Spark en el directorio raíz del usuario actual. Tenga en cuenta que para los escenarios de producción, no debe hacerlo.

Nota

Las versiones 2.7.7.0 y posteriores ya no instalan todas las dependencias de terceros necesarias. Las dependencias de terceros necesarias se agrupan en la carpeta third-party/lib en el archivo zip y se deben instalar manualmente.

Cree una instancia del servicio Compute. Para obtener instrucciones, consulte Creación de una instancia.
Asegúrese de que la instancia de servicio tenga una dirección IP pública para poder conectarse con una conexión de shell seguro (SSH). Para obtener instrucciones, consulte Conexión a una instancia.
Conéctese a su instancia de servicio mediante una conexión SSH.
Instale Spark y sus dependencias, Java y Scala, utilizando los ejemplos de código que aparecen a continuación.

# We'll use wget to download some of the artifacts that need to be installed
sudo yum install wget
 
# First install Java
sudo yum install java-1.8.0-openjdk.x86_64
export JAVA_HOME=/usr/lib/jvm/jre-1.8.0-openjdk
# Should be something like: OpenJDK Runtime Environment (build 1.8.0_161-b14)
java -version
 
# Then install Scala
wget https://downloads.lightbend.com/scala/2.12.4/scala-2.12.4.rpm
sudo yum install scala-2.12.4.rpm
# Should be something like: Scala code runner version 2.12.4 -- Copyright 2002-2017, LAMP/EPFL and Lightbend, Inc.
scala -version
 
# Then download Spark
wget https://archive.apache.org/dist/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz
tar xvf spark-2.2.1-bin-hadoop2.7.tgz
export SPARK_HOME=$HOME/spark-2.2.1-bin-hadoop2.7
export PATH=$PATH:$SPARK_HOME/bin
 
# Start a Spark master
cd $SPARK_HOME
./sbin/start-master.sh

Descarga del conector HDFS y creación de archivos de configuración

Nota

Para realizar este ejemplo, coloque los archivos JAR y de clave en el directorio raíz del usuario actual. Para los escenarios de producción, en su lugar, coloque estos archivos en un lugar común que aplique los permisos adecuados (es decir, que pueda leerlos el usuario con el que se están ejecutando Spark y Hive).

Descargue el conector HDFS en la instancia de servicio y agregue los archivos de configuración relevantes mediante el siguiente ejemplo de código. Para obtener más información, consulte Conector HDFS para almacenamiento de objetos.

wget https://github.com/oracle/oci-hdfs-connector/releases/download/v2.9.2.1/oci-hdfs.zip
unzip oci-hdfs.zip -d oci-hdfs
 
cd $HOME
mkdir .oci
# Create or copy your API key into the $HOME/.oci directory
 
cd $SPARK_HOME/conf
# Create a core-site.xml (e.g. by transferring one you have, using vi etc.). Consult
# https://docs.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#Properties
# for what this should look like
 
# Create a spark-defaults.conf file from the template
cp spark-defaults.conf.template spark-defaults.conf

En el archivo spark-defaults.conf, agregue lo siguiente en la parte inferior:

spark.sql.hive.metastore.sharedPrefixes= shaded.oracle,com.oracle.bmc

Preparación de datos

Para probar los datos, utilizaremos el juego de datos MovieLens.

Descargue el último juego de datos en https://grouplens.org/datasets/movielens/latest/. Asegúrese de descargar el juego de datos "pequeño".
Descomprima el archivo descargado.
Cargue el archivo movies.csv en el cubo de Object Storage.

Prueba con el shell de Spark

Con los datos listos, ahora podemos iniciar el shell de Spark y probarlo con un comando de ejemplo:

cd $SPARK_HOME
./bin/spark-shell
 
scala> sc.wholeTextFiles("oci://PipedUploadTest@sampletenancy/")
java.io.IOException: No FileSystem for scheme: oci

Se recibe un error en este punto porque el esquema del sistema de archivos oci:// no está disponible. Necesitamos hacer referencia al archivo JAR antes de iniciar el shell de Spark. A continuación se muestra un ejemplo de cómo hacerlo:

./bin/spark-shell --jars $HOME/oci-hdfs/lib/oci-hdfs-full-1.2.7.jar --driver-class-path $HOME/oci-hdfs/lib/oci-hdfs-full-1.2.7.jar
 
scala> sc.wholeTextFiles("oci://PipedUploadTest@sampletenancy/")
res0: org.apache.spark.rdd.RDD[(String, String)] = oci://PipedUploadTest@sampletenancy/ MapPartitionsRDD[1] at wholeTextFiles at <console>:25
 
scala> sc.textFile("oci://PipedUploadTest@sampletenancy/movies.csv").take(20).foreach(println)
movieId,title,genres
1,Toy Story (1995),Adventure|Animation|Children|Comedy|Fantasy
2,Jumanji (1995),Adventure|Children|Fantasy
3,Grumpier Old Men (1995),Comedy|Romance
4,Waiting to Exhale (1995),Comedy|Drama|Romance
5,Father of the Bride Part II (1995),Comedy
6,Heat (1995),Action|Crime|Thriller
7,Sabrina (1995),Comedy|Romance
8,Tom and Huck (1995),Adventure|Children
9,Sudden Death (1995),Action
10,GoldenEye (1995),Action|Adventure|Thriller
11,"American President, The (1995)",Comedy|Drama|Romance
12,Dracula: Dead and Loving It (1995),Comedy|Horror
13,Balto (1995),Adventure|Animation|Children
14,Nixon (1995),Drama
15,Cutthroat Island (1995),Action|Adventure|Romance
16,Casino (1995),Crime|Drama
17,Sense and Sensibility (1995),Drama|Romance
18,Four Rooms (1995),Comedy
19,Ace Ventura: When Nature Calls (1995),Comedy

El comando se ha realizado correctamente para que podamos conectarnos a Object Storage. Tenga en cuenta que si no desea transferir el argumento --jars cada vez que se ejecute el comando, puede copiar el archivo JAR oci-hdfs-full en el directorio $SPARK_HOME/jars.

Inicio del servidor Thrift de Spark

Inicie el servidor Thrift de Spark en el puerto 10015 y utilice la herramienta de línea de comandos Beeline para establecer una conexión JDBC y, a continuación, ejecutar una consulta básica, como se muestra a continuación:

cd $SPARK_HOME
./sbin/start-thriftserver.sh --hiveconf hive.server2.thrift.port=10015

Una vez que el servidor de Spark se está ejecutando, podemos iniciar Beeline, como se muestra aquí:

cd $SPARK_HOME
./bin/beeline
Beeline version 1.2.1.spark2 by Apache Hive
beeline>

A continuación, conéctese al servidor, como se muestra aquí:

Nota

Para realizar este ejemplo, no hemos configurado ninguna seguridad, por lo que se aceptará cualquier nombre de usuario y contraseña. Para escenarios de producción, no debe hacerlo.

beeline> !connect jdbc:hive2://localhost:10015 testuser testpass
Connecting to jdbc:hive2://localhost:10015
log4j:WARN No appenders could be found for logger (org.apache.hive.jdbc.Utils).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Connected to: Spark SQL (version 2.2.1)
Driver: Hive JDBC (version 1.2.1.spark2)
Transaction isolation: TRANSACTION_REPEATABLE_READ
0: jdbc:hive2://localhost:10015>

Si ahora comprobamos qué tablas existen, vemos lo siguiente:

0: jdbc:hive2://localhost:10015> show tables;
+-----------+------------+--------------+--+
| database  | tableName  | isTemporary  |
+-----------+------------+--------------+--+
+-----------+------------+--------------+--+
No rows selected (0.724 seconds)

Ninguna existe en este momento; sin embargo, podemos crear una tabla y enlazarla al archivo movies.csv que descargó y colocó en el cubo de Object Storage, como se muestra aquí:

0: jdbc:hive2://localhost:10015> create table test_table (movieId integer, title string, genres string) using csv options (path "oci://myBucket@myTenant/movies.csv", header "true", delimiter ",");
 
0: jdbc:hive2://localhost:10015> describe formatted test_table;
+-------------------------------+------------------------------------------------------------+----------+--+
|           col_name            |                         data_type                          | comment  |
+-------------------------------+------------------------------------------------------------+----------+--+
| movieId                       | int                                                        | NULL     |
| title                         | string                                                     | NULL     |
| genres                        | string                                                     | NULL     |
|                               |                                                            |          |
| # Detailed Table Information  |                                                            |          |
| Database                      | default                                                    |          |
| Table                         | test_table                                                 |          |
| Owner                         | opc                                                        |          |
| Created                       | Thu Mar 01 20:45:18 GMT 2018                               |          |
| Last Access                   | Thu Jan 01 00:00:00 GMT 1970                               |          |
| Type                          | EXTERNAL                                                   |          |
| Provider                      | csv                                                        |          |
| Table Properties              | [transient_lastDdlTime=1519937118]                         |          |
| Location                      | oci://PipedUploadTest@sampletenancy/movies.csv    |          |
| Serde Library                 | org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe         |          |
| InputFormat                   | org.apache.hadoop.mapred.SequenceFileInputFormat           |          |
| OutputFormat                  | org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat  |          |
| Storage Properties            | [delimiter=,, header=true, serialization.format=1]         |          |
+-------------------------------+------------------------------------------------------------+----------+--+

Tenga en cuenta que la tabla almacena sus datos de forma externa en Object Storage y que se puede acceder a ellos mediante el conector HDFS (esquema del sistema de archivos oci://). Ahora que tenemos una tabla, podemos consultarla:

0: jdbc:hive2://localhost:10015> select * from test_table limit 10;
+----------+-------------------------------------+----------------------------------------------+--+
| movieId  |                title                |                    genres                    |
+----------+-------------------------------------+----------------------------------------------+--+
| 1        | Toy Story (1995)                    | Adventure|Animation|Children|Comedy|Fantasy  |
| 2        | Jumanji (1995)                      | Adventure|Children|Fantasy                   |
| 3        | Grumpier Old Men (1995)             | Comedy|Romance                               |
| 4        | Waiting to Exhale (1995)            | Comedy|Drama|Romance                         |
| 5        | Father of the Bride Part II (1995)  | Comedy                                       |
| 6        | Heat (1995)                         | Action|Crime|Thriller                        |
| 7        | Sabrina (1995)                      | Comedy|Romance                               |
| 8        | Tom and Huck (1995)                 | Adventure|Children                           |
| 9        | Sudden Death (1995)                 | Action                                       |
| 10       | GoldenEye (1995)                    | Action|Adventure|Thriller                    |
+----------+-------------------------------------+----------------------------------------------+--+

Documentación de Oracle Cloud Infrastructure