Especificando programaticamente o esquema

O segundo método para criar DataFrame é por meio de uma interface programática que permite construir um esquema e aplicá-lo a um RDD existente. Podemos criar um DataFrame programaticamente usando as três etapas a seguir.

Crie um RDD de linhas a partir de um RDD original.
Crie o esquema representado por um StructType correspondendo à estrutura de Linhas no RDD criado na Etapa 1.
Aplique o esquema ao RDD de linhas por meio do método createDataFrame fornecido por SQLContext.

Exemplo

Vamos considerar um exemplo de registros de funcionários em um arquivo de texto chamado employee.txt. Crie um Schema usando DataFrame diretamente lendo os dados do arquivo de texto.

Given Data - Veja os seguintes dados de um arquivo chamado employee.txt colocado no diretório respectivo atual em que o ponto do shell do spark está em execução.

1201, satish, 25
1202, krishna, 28
1203, amith, 39
1204, javed, 23
1205, prudvi, 23

Siga as etapas fornecidas a seguir para gerar um esquema programaticamente.

Abra o Spark Shell

Inicie o shell do Spark usando o exemplo a seguir.

$ spark-shell

Criar objeto SQLContext

Gere SQLContext usando o seguinte comando. Aqui,sc significa objeto SparkContext.

scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)

Ler entrada de arquivo de texto

Crie um DataFrame RDD lendo os dados do arquivo de texto chamado employee.txt usando o seguinte comando.

scala> val employee = sc.textFile("employee.txt")

Crie um esquema codificado em um formato de string

Use o seguinte comando para criar um esquema codificado em um formato de string. Ou seja, assuma a estrutura de campo de uma tabela e passe os nomes dos campos usando algum delimitador.

scala> val schemaString = "id name age"

Resultado

schemaString: String = id name age

Importar APIs respectivas

Use o seguinte comando para importar recursos de linha e DataTypes SQL.

scala> import org.apache.spark.sql.Row;
scala> import org.apache.spark.sql.types.{StructType, StructField, StringType};

Gerar Esquema

O seguinte comando é usado para gerar um esquema lendo o schemaStringvariável. Isso significa que você precisa ler cada campo dividindo toda a string com um espaço como delimitador e considerar cada tipo de campo como String, por padrão.

scala> val schema = StructType(schemaString.split(" ").map(fieldName ⇒ StructField(fieldName, StringType, true)))

Aplicar transformação para ler dados de arquivo de texto

Use o seguinte comando para converter um RDD (funcionário) em linhas. Isso significa que aqui estamos especificando a lógica para ler os dados RDD e armazená-los no rowRDD. Aqui, estamos usando duas funções de mapa: uma é um delimitador para dividir a string de registro (.map(_.split(","))) e a segunda função de mapa para definir uma Linha com o valor de índice do campo (.map(e ⇒ Row(e(0).trim.toInt, e(1), e(2).trim.toInt)))

scala> val rowRDD = employee.map(_.split(",")).map(e ⇒ Row(e(0).trim.toInt, e(1), e(2).trim.toInt))

Aplicar RowRDD em dados de linha com base no esquema

Use a seguinte instrução para criar um DataFrame usando rowRDD dados e schema (SCHEMA) variável.

scala> val employeeDF = sqlContext.createDataFrame(rowRDD, schema)

Resultado

employeeDF: org.apache.spark.sql.DataFrame = [id: string, name: string, age: string]

Armazenar dados DataFrame na tabela

Use o seguinte comando para armazenar o DataFrame em uma tabela chamada employee.

scala> employeeDF.registerTempTable("employee")

o employeea mesa agora está pronta. Vamos passar algumas consultas SQL para a tabela usando o métodoSQLContext.sql().

Selecione Consulta no DataFrame

Use a seguinte declaração para selecionar todos os registros do employeemesa. Aqui usamos a variávelallrecordspara capturar todos os dados de registros. Para exibir esses registros, ligueshow() método sobre ele.

scala> val allrecords = sqlContext.sql("SELECT * FROM employee")

Para ver os dados do resultado de allrecords DataFrame, use o seguinte comando.

scala> allrecords.show()

Resultado

+------+--------+----+
|  id  | name   |age |
+------+--------+----+
| 1201 | satish | 25 |
| 1202 | krishna| 28 |
| 1203 | amith  | 39 |
| 1204 | javed  | 23 |
| 1205 | prudvi | 23 |
+------+--------+----+

O método sqlContext.sqlpermite que você construa DataFrames quando as colunas e seus tipos não são conhecidos até o tempo de execução. Agora você pode executar diferentes consultas SQL nele.

↰ Previous page