Thursday, November 5, 2020

Spark Create DataFrame with Examples

In Spark,to create dataset 1. createDataFrame() and 2. toDF() methods These methods you can create a Spark DataFrame from already existing RDD, DataFrame, Dataset, List, Seq data objects.. In Spark, createDataFrame() and toDF() methods are used to create a DataFrame, using these methods you can create a Spark DataFrame from already existing RDD, DataFrame, Dataset, List, Seq data objects. 1. Spark Create DataFrame from RDD import spark.implicits._ val columns = Seq("language","users_count") val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000")) val rdd = spark.sparkContext.parallelize(data) 1.1 Using toDF() function val dfFromRDD1 = rdd.toDF() dfFromRDD1.printSchema() o/p: default it creates with _1 and _2 and so on for column names root |-- _1: string (nullable = true) |-- _2: string (nullable = true) val dfFromRDD1 = rdd.toDF("language","users_count") dfFromRDD1.printSchema() o/p: root |-- language: string (nullable = true) |-- users: string (nullable = true) 1.2 Using Spark createDataFrame() from SparkSession val dfFromRDD2 = spark.createDataFrame(rdd).toDF(columns:_*) 1.3 Using createDataFrame() with the Row type val schema = StructType( Array(StructField("language", StringType,true), StructField("language", StringType,true))) val rowRDD = rdd.map(attributes => Row(attributes._1, attributes._2)) val dfFromRDD3 = spark.createDataFrame(rowRDD,schema) 3. Creating Spark DataFrame from CSV Here, will see how to create from a CSV file. val df2 = spark.read.csv("/src/resources/file.csv") 4. Creating from text (TXT) file Here, will see how to create from a TXT file. val df2 = spark.read .text("/src/resources/file.txt") 5. Creating from JSON file Here, will see how to create from a JSON file. val df2 = spark.read .json("/src/resources/file.json") 6. Creating from an XML file To create DataFrame by parse XML, we should use DataSource "com.databricks.spark.xml" spark-xml api from Databricks. com.databricks spark-xml_2.11 0.6.0 val df = spark.read .format("com.databricks.spark.xml") .option("rowTag", "person") .xml("src/main/resources/persons.xml") 7. Creating from Hive val hiveContext = new org.apache.spark.sql.hive.HiveContext(spark.sparkContext) val hiveDF = hiveContext.sql(“select * from emp”) 8. Creating from the Database table (RDBMS) 8.a) From Mysql table Make sure you have MySQL library as a dependency in your pom.xml file or MySQL jars in your classpath. val df_mysql = spark.read.format(“jdbc”) .option(“url”, “jdbc:mysql://localhost:port/db”) .option(“driver”, “com.mysql.jdbc.Driver”) .option(“dbtable”, “tablename”) .option(“user”, “user”) .option(“password”, “password”) .load() 8.1 From DB2 table Make sure you have DB2 library as a dependency in your pom.xml file or DB2 jars in your classpath. val df_db2 = spark.read.format(“jdbc”) .option(“url”, “jdbc:db2://localhost:50000/dbname”) .option(“driver”, “com.ibm.db2.jcc.DB2Driver”) .option(“dbtable”, “tablename”) .option(“user”, “user”) .option(“password”, “password”) .load() Similarly, we can create DataFrame in Spark from most of the relational databases which I’ve not covered here and I will leave this to you to explore. 9. Create DataFrame from HBase table To create Spark DataFrame from the HBase table, we should use DataSource defined in Spark HBase connectors. for example use DataSource “org.apache.spark.sql.execution.datasources.hbase” from Hortonworks or use “org.apache.hadoop.hbase.spark“from spark HBase connector. val hbaseDF = sparkSession.read .options(Map(HBaseTableCatalog.tableCatalog -> catalog)) .format("org.apache.spark.sql.execution.datasources.hbase") .load() Detail example explained at Generating DataFrame from HBase table

Spark DF Join

cat input.txt transaction_id,user_name,user_type,transaction_date 1,user1,New,20-04-2018 2,user2,Privileged,20-11-2019 3,user3,Privileged,20-04-2018 4,user4,New,22-05-2019 5,user5,New,20-04-2019 6,user6,New,25-06-2018 7,user7,New,20-04-2018 8,user8,Privileged,20-04-2019 9,user9,Privileged,20-04-2018 10,user10,New,20-04-2019 cat vailadte.csv val_date 2019 import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.SparkSession object SparkDFJoin { def main(args: Array[String]): Unit = { Logger.getLogger("org").setLevel(Level.OFF) val spark : SparkSession = SparkSession.builder().appName("SparkJoinDf").master("local[1]").getOrCreate() import spark.implicits._ val df1 = spark.read.format("csv") .option("header","true") .option("inferSchema","true") .load("C:\\Users\\shrav\\IdeaProjects\\test\\src\\main\\scala\\input.csv") val addcol = df1.withColumn("year",$"transaction_date".substr(7,4)) addcol.createOrReplaceTempView("userdata") val df2= spark.read.format("csv").option("header","true").option("inferschema","true") .load("C:\\Users\\shrav\\IdeaProjects\\test\\src\\main\\scala\\vailadtedate.csv") df2.createOrReplaceTempView("valtable") //spark.sqlContext.sql("select * from userdata a left join valtable b ON a.year=b.val_date").show() /*+--------------+---------+----------+----------------+----+--------+ |transaction_id|user_name| user_type|transaction_date|year|val_date| +--------------+---------+----------+----------------+----+--------+ | 1| user1| New| 20-04-2018|2018| null| | 2| user2|Privileged| 20-11-2019|2019| 2019| | 3| user3|Privileged| 20-04-2018|2018| null| | 4| user4| New| 22-05-2019|2019| 2019| | 5| user5| New| 20-04-2019|2019| 2019| | 6| user6| New| 25-06-2018|2018| null| | 7| user7| New| 20-04-2018|2018| null| | 8| user8|Privileged| 20-04-2019|2019| 2019| | 9| user9|Privileged| 20-04-2018|2018| null| | 10| user10| New| 20-04-2019|2019| 2019| +--------------+---------+----------+----------------+----+--------+ */ spark.sqlContext.sql("select transaction_id,user_name,user_type,transaction_date from " + "(select * from userdata a left join valtable b ON a.year=b.val_date) where val_date Is not NUll").show() /* +--------------+---------+----------+----------------+ |transaction_id|user_name| user_type|transaction_date| +--------------+---------+----------+----------------+ | 2| user2|Privileged| 20-11-2019| | 4| user4| New| 22-05-2019| | 5| user5| New| 20-04-2019| | 8| user8|Privileged| 20-04-2019| | 10| user10| New| 20-04-2019| +--------------+---------+----------+----------------+ */ } }

Friday, October 30, 2020

hive Parquet to textfile

Insert overrwrite Directory '/user/test' row format delimited fileds terminated by 'u0001' storted as textfile select nvl(id, ""), nvl(asd, "") From db.test

Sunday, October 18, 2020

hive example query

Create database if not exist testdb location /user/testdb.db';

Drop table if exists testdb.emp;

Create table testdb.emp row format delimited fields terminated by '~' escaped by '\\134' lines terminated by '\n' null defined as ' ' stored as textfile as select case when ascii('empid') = 0 and length('empid) > 0 then " " else 'empid' end 'empid', concat(nvl ('empapproved', ' ' , "@#$") From xyzdb.abc where empId <> 1234;

Wednesday, October 7, 2020

sqoop Compression snappy and avro Format

locate core-site.xml -> find codec type available

Example of loading data from MySQL to HDFS (compression: Snappy and Avro format)

$ sqoop import \
 --connect jdbc:mysql://localhost:33/mybbdd \
 --username=root -P \
 --table=mytable \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_snappy_avro \
 --compress \
 --compression-codec org.apache.hadoop.io.compress.SnappyCodec \ 
 --as-avrodatafile

Example of loading data from MySQL to HDFS (compression: gzip and Avro format)

$ sqoop import \
 --connect jdbc:mysql://localhost/mibbdd \
 --username=root -P \
 --table=mitabla \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_gzip_avro \
 --compress \
 --compression-codec org.apache.hadoop.io.compress.GzipCodec \
 --as-avrodatafile

Example of loading data from MySQL to HDFS (compression: BZIP2 and Sequence format)

$ sqoop import \
 --connect jdbc:mysql://localhost/mibbdd \
 --username=root -P \
 --table=mitabla \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_bzip2_sequence \
 --compress \
 --compression-codec org.apache.hadoop.io.compress.BZip2Codec \
 --as-sequencefile

Example of loading data from MySQL to HDFS (restricting data with columns)

$ sqoop import \
 --connect jdbc:mysql://localhost/mibbdd \
 --username=root -P \
 --table=mitabla \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_2_columns \
 --columns nombre,edad

Example of loading data from MySQL to HDFS (restricting data with WHERE)

$ sqoop import \
 --connect jdbc:mysql://localhost/mybbdd \
 --username=root -P \
 --table=mytable \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_mayor_age_40 \
 --where "edad > 40"

Example of loading data from MySQL to HDFS (incremental load)

In order to make an incremental insertion we need to include new data to the table “MyTable”, for this we execute in MySQL the following sentence:

mysql> 
INSERT INTO mytable (nombre, edad, salario) VALUES
        ("Diego", 24, 21000), ("Rosa", 26, 24000), ("Javier", 28, 25000), ("Lorena", 35, 28000), ("Miriam", 42, 30000), ("Patricia", 43, 25000), ("Natalia", 45, 39000);

Note: To make the insertion necessary to do it in the db “MIBBDD”

Once the insertion is done we can make the incremental insertion from the 8 as it is the first element introduced in the new insertion.

$ sqoop import \ 
 --connect jdbc:mysql://localhost/mybbdd \ 
 --username=root -P \
 --table=mytable \ 
 --driver=com.mysql.jdbc.Driver \ 
 --target-dir=/my_table_hdfs \ 
 --incremental append \
 --check-column id \
 --last-value 8

Example of loading data from MySQL to HDFS and consultable from HIVE

In order to make an insertion of the table in the hive database, we must create db where it will be inserted, to avoid problems:

Hive > CREATE DATABASE mybbddhive;

Once the database is created, you are ready to run the query:

$ sqoop import \
 --connect jdbc:mysql://localhost/mybbdd \
 --username=root -P \
 --table=mytable \
 --driver=com.mysql.jdbc.Driver \
 --target-dir=/ej_hive \
 --compress \
 --compression-codec org.apache.hadoop.io.compress.SnappyCodec \ 
 --hive-import \
 --hive-database mihive \ 
 --create-hive-table \
 --hive-table ej_hive_table

Creating a MySQL database table
A database is created with a table on which to perform the tests, the following commands will be used.
accessing MYSQL
$ mysql
You can make an ERROR 1045 (28000): Access denied for user ‘ root ‘ @ ‘ localhost, which is resolved:
mysql -u root-P
Note: Keep in mind that the MySQL and system wash can be different.
Consult database
mysql> show databases;
Create Database
mysql> create database myddbb;
Use database
mysql> use myddbb;
Create base table
mysql> CREATE TABLE mytable (
         id MEDIUMINT NOT NULL AUTO_INCREMENT,
         name CHAR (30) NOT NULL,
         age INTEGER (30),
         salary INTEGER (30),
         PRIMARY KEY (id));
CREATE TABLE 2
mysql> CREATE TABLE mytable2 (
         id MEDIUMINT NOT NULL AUTO_INCREMENT,
         name CHAR (30) NOT NULL,
         age INTEGER (30),
         salary INTEGER (30),
         PRIMARY KEY (id));
Insert Data
mysql> INSERT INTO mytable (name, age, salary) values
        ("Peter", 24, 21000), ("Maria", 26, 24000), ("John", 28, 25000), ("Louis", 35, 28000), ("Monica", 42, 30000), ("Rose", 43, 25000), ("Susana", 45, 39000);
Note: If you have permission problems enter MySQL and give all permissions:
grant all privileges on *.* to 'root'@'localhost' IDENTIFIED BY 'MySQL_Key' WITH GRANT OPTION;
Exit MYSQL
mysql> exit;
 
Load MySQL data to HDFS
Example of loading data from the table “MyTable” of the Database “MIBBDD” to the folder HDFs name “Mitabla_hdfs”
$ sqoop Import \
 --connect jdbc:mysql://localhost/myddbb \
 --username = root -P \
 --table = mytable \
 --driver = com.mysql.jdbc.driver \
 --target-dir =/my_hdfs_table \
 --fields-terminated-by = ',' \
 --lines-terminated-by '\n'
Target-dir: File HDFS where it is stored.
Table: Identifies the table to be copied.
Clear-Staging-table: Indicates that past data can be deleted.
VERBOSE: Prints additional information to facilitate debugging.
Fields-terminated-by: defining the delimiter.
 
Loading data from HDFS to MySQL
Example of loading data from the HDFS folder named “my_hdfs_table” to the “mytable2” table in the “myddbb” database.

$ sqoop export 
 --connect jdbc: mysql://localhost/myddbb 
 --username = root -P 
 --table = mytable2 
 --export-dir =/my_hdfs_table -m 1
Note: If you have problems “set $ ACCUMULO_HOME to the root in your ACCUMULO intallation”, it can be avoided with:
$ ACCUMULO_HOME = '/var/lib/accumulo'
$ Export ACCUMULO_HOME
$ sudo mkdir/var/lib/accumulo
To practice with Sqoop then consult: “Examples of Sqoop”, in this section compiled many examples that can serve you useful.
Source: official Shell documentation
Source: official documentation for JAVA API

optimization tecqiue

WITH MAX_DATE AS (SELECT MAX(DATE) AS MAX_DATE FROM DEMODB.EMP) SELECT * FROM DEMODB.EMP A, MAX_DT B WHERE A.DATE=B.MAX_DATE;

Friday, October 2, 2020

saprk 2.0 jdbc

Spark-shell does not encounter so many problems This is a sbt dependency problem in IDEA.

1, import package problem

import java.util.Properties
import org.apache.spark.sql
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

The build.sbt file is as follows:

name := "Simple Project"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "2.1.0"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.1.2"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.1.2"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.1.2"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.2.0"
libraryDependencies += "mysql" % "mysql-connector-java" % "8.0.15"

The above org.apache.spark requires a package of 2.0.0 or higher. Otherwise, SparkSession cannot be imported.

Another puzzling thing is that the mysql-connector-java-8.0.15.jar imported from the outside of the dependency does not work, resulting in the package can not find the com.mysql.jdbc driver.

So: libraryDependencies += "mysql" % "mysql-connector-java" % "8.0.15" is to solve the driver problem.

Wait for sbt:dump to complete and run the code successfully.

--------------------------------

import java.util.Properties

import org.apache.spark.sql

import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

import org.apache.spark.sql.SparkSession

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

//import com.mysql.jdbc

//SparkSession

object ConnectJDBC {

def main(args: Array[String]): Unit = {

val conf = new SparkConf().setAppName("ConnectJDBC").setMaster("local[*]")

val sc = new SparkContext(conf)

val spark = SparkSession.builder().getOrCreate()

import spark.implicits._

// read the information

Val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/spark") //*****This is the database name

.option("driver", "com.mysql.jdbc.Driver").option("dbtable", "student")//***** is the table name

.option("user", "root").option("password", "123456").load()

jdbcDF.show()

/ / Below we set two data to represent two student information

val studentRDD = spark.sparkContext.parallelize(Array("1 Licheng M 26", "2 Jianghua M 27")).map(_.split(" "))

/ / The following to set the mode information

val schema = StructType(List(StructField("id", IntegerType, true), StructField("name", StringType, true), StructField("gender", StringType, true), StructField("age", IntegerType, true)))

/ / Create a Row object below, each Row object is a row in the rowRDD

val rowRDD = studentRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).trim, p(3).toInt))

/ / Establish a correspondence between the Row object and the mode, that is, the data and the pattern are associated

val studentDF = spark.createDataFrame(rowRDD, schema)

/ / Create a prop variable to save JDBC connection parameters

val prop = new Properties()

Prop.put("user", "root") // indicates that the username is root

Prop.put("password", "123456") // indicates that the password is hadoop

Prop.put("driver", "com.mysql.jdbc.Driver") // indicates that the driver is com.mysql.jdbc.Driver

// /usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar

/ / The following can be connected to the database, using append mode, indicating additional records to the student table in the database spark

studentDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/spark", "spark.student", prop)

}

---------------------------

operation result:

+---+---------+------+---+
| id| name|gender|age|
+---+---------+------+---+
| 1| Licheng| M| 26|
| 2| Jianghua| M| 27|
+---+---------+------+---+