Apache Drill JDBC Example

package packagename;

import java.sql.*;

public class DrillJDBCExample1 {
    static final String JDBC_DRIVER = "org.apache.drill.jdbc.Driver";
    static final String DB_URL = "jdbc:drill:zk=machineName:5181/drill/x-drillbits";

//You can get this URL from drill explorer, if you have installed
//Zookeeper port is 5181 in MapR hadoop


    //static final String USER = "admin";
    //static final String PASS = "admin";

    public static void main(String[] args) {
        Connection conn = null;
        Statement stmt = null;
        try{

        Class.forName(JDBC_DRIVER);
            conn = DriverManager.getConnection(DB_URL," "," ");
            stmt = conn.createStatement();

            /* Perform a select on data in the classpath storage plugin. */
         
            String sql = "select transaction_week_end,SUM(SumOfSale) as sq,SUM(SumOfPrice) as sp from dfs.`StoragepluginName`.`drillTableName` GROUP BY `transaction_week_end`";
         
            ResultSet rs = stmt.executeQuery(sql);

            while(rs.next()) {
            
             System.out.print(rs.getBigDecimal("sq")+"\t");
            System.out.print(rs.getBigDecimal("sp")+"\n");
            
            }

            rs.close();
            stmt.close();
            conn.close();
        } catch(SQLException se) {
            //Handle errors for JDBC
            se.printStackTrace();
        } catch(Exception e) {
            //Handle errors for Class.forName
            e.printStackTrace();
        } finally {
            try{
                if(stmt!=null)
                    stmt.close();
            } catch(SQLException se2) {
            }
            try {
                if(conn!=null)
                    conn.close();
            } catch(SQLException se) {
                se.printStackTrace();
            }
        }
    }
}

How to read a Parquet file and make a dataframe and create Hive temp table

package packagename;



import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.hive.thriftserver.*;



public class SparkReadParquetAndRegTempTable {


public static void main(String[] args) throws ClassNotFoundException {


SparkConf spconf = new SparkConf();

spconf.set("spark.driver.maxResultSize", "3g");

JavaSparkContext sc=null;
try
{
sc = new JavaSparkContext(spconf);

//SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);

HiveContext sqlContext = new org.apache.spark.sql.hive.HiveContext(sc.sc());


DataFrame df = sqlContext.read().parquet("/path/parquetFolderName.parquet");


df.printSchema();


//To Query the table via beeline as Spark Hive table

df.registerTempTable("tempTable_spark");

HiveThriftServer2.startWithContext(sqlContext);

}
catch(Exception e)
{
System.out.print("Error is"+e.toString());
}



}

 


}


How to submit the job

hadoop_classpath=$(hadoop classpath)
HBASE_CLASSPATH=$(hbase classpath)

sudo -u userName /spark/spark-1.5.2/bin/spark-submit   --name tempSparkTable     --class packageName.SparkReadParquetAndRegTempTable  --master local[4]   --num-executors 8    --executor-cores 8    --executor-memory 8G   --conf "spark.executor.extraClassPath=${HBASE_CLASSPATH}"   --conf "spark.driver.extraClassPath=${HBASE_CLASSPATH}"    --conf "spark.executor.extraClassPath=${hadoop_classpath}"
 --jars /path/projectName-0.0.1-SNAPSHOT-jar-with-dependencies.jar
/path/projectName-0.0.1-SNAPSHOT.jar

Make sure, there is no Hive Thrift server running in port 10000 in the machine, where you run this program


Connect via /opt/mapr/hive/hive-1.2/bin/beeline -u jdbc:hive2://serverName:10000 -n UserName

Show tables;

should list your table name in the list

tempTable_spark

Then you can run the queries against this temp table


Maven Dependencies:

<dependencies>

  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

 <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

 <dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-common</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-examples</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>9.4.1208</version>
</dependency>

<dependency>
    <groupId>com.databricks</groupId>
    <artifactId>spark-csv_2.11</artifactId>
    <version>1.2.0</version>
</dependency>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive-thriftserver_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

</dependencies>

How to create Spark Dataframe from (Read) PostgreSql and write processed data frame to PostgreSql/MySql

package com.packagename;


import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

public class SparkReadFromPostgres {

public static void main(String[] args) {

Map<String, String> options = new HashMap<String, String>();

options.put("url", "jdbc:postgresql://servername:5432/dbname");
options.put("user", "username");
options.put("password", "<somePassword>");
options.put("driverClassName", "org.postgresql.Driver");
options.put("dbtable", "schema.tableName");

SparkConf spconf = new SparkConf();

spconf.set("spark.driver.maxResultSize", "3g");

JavaSparkContext sc=null;

try
{
sc = new JavaSparkContext(spconf);

SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);

DataFrame jdbcDF = sqlContext.read().format("jdbc").options(options).load();

jdbcDF.printSchema();

jdbcDF.show(10);


//if you want to create the table again in Postgres

createPostGresTable(jdbcDF,"testTable");

}
finally
{
sc.close();
}

}

public static void createPostGresTable(DataFrame output, String postgresTableName)
{
String url = "jdbc:postgresql://serverName:5432/dbName";
Properties props = new Properties();
props.setProperty("user","userName");
props.setProperty("password","password");
//props.setProperty("ssl","true");
props.setProperty("driverClassName","org.postgresql.Driver");

//Connection conn = DriverManager.getConnection(url, props);

//String postgresTable="TableName";

output.write().mode("overwrite").jdbc(url, postgresTableName, props);

}


}

Command to Run the Spark Program

hadoop_classpath=$(hadoop classpath)
HBASE_CLASSPATH=$(hbase classpath)


sudo -u userId  /spark/spark-1.5.2/bin/spark-submit   --name SparkReadFromPostgres     --class com.packagename.SparkReadFromPostgres   --master yarn   --deploy-mode client   --num-executors 8    --executor-cores 8    --executor-memory 4G   --conf "spark.executor.extraClassPath=${HBASE_CLASSPATH}"   --conf "spark.driver.extraClassPath=${HBASE_CLASSPATH}"    --conf "spark.executor.extraClassPath=${hadoop_classpath}"  --conf "spark.executor.extraClassPath=/sharedpath/postgresql-9.4.1208.jar"   --conf "spark.driver.extraClassPath=/sharedpath/postgresql-9.4.1208.jar"  --jars /projectName-0.0.1-SNAPSHOT-jar-with-dependencies.jar  /projectName-0.0.1-SNAPSHOT.jar

pom.xml dependencies:

<dependencies>

  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

 <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

 <dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-common</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-examples</artifactId>
    <version>0.98.12-hadoop2</version>
</dependency>

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>9.4.1208</version>
</dependency>

<dependency>
    <groupId>com.databricks</groupId>
    <artifactId>spark-csv_2.11</artifactId>
    <version>1.2.0</version>
</dependency>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive-thriftserver_2.10</artifactId>
    <version>1.5.2</version>
</dependency>

</dependencies>