Model Data Anylysis Using DataFrame and SparkSQL in java

-

Here we will analyze Model data using pure Spark SQL, Data Frame and will use mostly used methods with sample data

package com.dpq.model.data.driver;

import java.util.Arrays;

import org.apache.spark.SparkConf;

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

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

public class ModelDataAnalysis {

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

JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName(“Spark Count”).setMaster(“local”));

SparkSession spark = SparkSession.builder().appName(“spark-bigquery-demo”).getOrCreate();

Dataset<Row> row = spark.read().csv(“/Users/dpq/springbootWrokspace/CountryDataAnalysis/resources/modeloutput.csv”);

// way 1 to change column name

row = row.withColumnRenamed(“_c0”, “CountryName”);

row = row.withColumnRenamed(“_c1”, “ReportingPurpuse”);

row = row.withColumnRenamed(“_c2”, “Quarter”);

row = row.withColumnRenamed(“_c3”, “Grain”);

row = row.withColumnRenamed(“_c4”, “PP0”);

row = row.withColumnRenamed(“_c5”, “PP1”);

row = row.persist();

row.show();

Arrays.stream(row.columns()).forEach(column -> System.out.println(column));

/**Seq<String> sourceColumns = JavaConverters.asScalaBuffer( Arrays.asList(row.columns()) );

Column [] arr = {new Column(“COUNTRY”),new Column(“REPORTING_PERIOD”),

new Column(“QUATER”), new Column(“GRAIN”), new Column(“PROJECTED_1ST_MONTH”),

new Column(“PROJECTED_2ND_MONTH”)};

Seq<Column> destinationColumns = JavaConverters.asScalaBuffer( Arrays.asList( arr) );

// way 2 to change ALL column name

row = row.withColumn(sourceColumns , destinationColumns); **/

row.dropDuplicates().show();

System.out.println(row.count());

System.out.println(row.dropDuplicates().count());

   

//removing duplicates

row = row.dropDuplicates();

//creating table Model_Output on top of file which loaded

row.createOrReplaceTempView(“Model_Output”);

// way 2 to change column nameS

Dataset<Row> sqlDF = spark.sql(“SELECT CountryName as COUNTRY, ReportingPurpuse AS RPT_PRD, “

+ “Quarter AS QUATER, Grain AS GRAIN, PP0 AS PRE_PROJECTED_1ST_MONTH,PP1 AS PRE_PROJECTED_2ND_MONTH FROM Model_Output”);

//1ST way to filter records

Dataset<Row> filteredRecords1=sqlDF.filter(sqlDF.col(“QUATER”).isInCollection(Arrays.asList(“2017Q1”, “2018Q1”, “2019Q1”)));

filteredRecords1.show();

//2nd way to filter records

sqlDF = spark.sql(“SELECT CountryName as COUNTRY, ReportingPurpuse AS RPT_PRD, “

+ “Quarter AS QUATER, Grain AS GRAIN, PP0 AS PRE_PROJECTED_1ST_MONTH,PP1 AS PRE_PROJECTED_2ND_MONTH “

+ “FROM Model_Output where Quarter not in (\”2016Q1\”)”);

sqlDF.show();

Dataset<Row> finalAggregatedResults = spark.sql(“SELECT CountryName as COUNTRY, ReportingPurpuse AS RPT_PRD, “

+ “Quarter AS QUATER, Grain AS GRAIN, sum(CAST(PP0 AS DOUBLE)) ,sum(CAST(PP1 AS DOUBLE))  “

+ “FROM Model_Output group by CountryName,ReportingPurpuse,Quarter,Grain”);

finalAggregatedResults.show();

sc.close();

}

Refer GIT URL for code: https://github.com/dpq1422/ModelDataAnalysis

 

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more