External table access via spark partition drop

For the external table in hive if you are adding or dropping partitions and if you try to access the table it will give vertex failure issue.

MSCK will not fix it
MSCK REPAIR TABLE basan.shipment_ext;

To fix this drop the partitions.

show partitions basan.shipment_ext;
//dropping single partition
ALTER TABLE basan.shipment_ext DROP IF EXISTS PARTITION(order_date ='2019-07-15')
//dropping multiple partition
ALTER TABLE basan.shipment_ext DROP IF EXISTS PARTITION(order_date >'2018-07-15')

Spark with mysql database(spark jdbc)

/install the mysql server 8.0.19 and create the schema basan and the table shipment

CREATE TABLE `shipment` (
  `order_number` text,
  `order_date` text,
  `ship_status` text
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;


  1.make sure jar version is mysql-connector-java-8.0.19.jar and spark version in 2.3.2
  2. copy mysql-connector-java-8.0.19.jar jar to the location /jars
  datanucleus-core-3.2.10.jar
  3. run spark-shell and runt he below commands to interact with jdbc
 
 val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/basan?useSSL=false")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "shipment")
  .option("user", "root")
  .option("password", "root1234")
  .load()

import spark.implicits._

val someDF = Seq(
  ("ordernumberinsert1", "2020-02-05", "delivered"),
  ("ordernumberinsert2", "2020-02-05", "delivered"),
  ("ordernumberinsert3", "2020-02-05", "delivered")
).toDF("order_number", "order_date","ship_status")
someDF.write.mode("overwrite")
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/basan?useSSL=false")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "shipment")
  .option("user", "root")
  .option("password", "root1234")
  .save()

Filtering set of values from the bigger dataframe

Filtering set of values from the bigger dataframe

import spark.implicits._
    val localFileDF = spark.read.orc("/tmp/Analysis-10")
        localFileDF.filter(localFileDF("key") === "1065551413044" )



var followinglist=List("1065551413044")
val df2 = localFileDF.filter(col("key").isin(followinglist:_*))

back to basic- sql left join producing duplicate records

CREATE TABLE `shipment` (
  `shipmentnumber` varchar(45) DEFAULT NULL,
  `order_number` varchar(45) DEFAULT NULL,
  `order_line` varchar(45) DEFAULT NULL,
  `quantity` int DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

CREATE TABLE `order` (

  `order_number` varchar(45) DEFAULT NULL,

  `order_line` varchar(45) DEFAULT NULL,

  `prder_type` varchar(45) DEFAULT NULL

) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

INSERT INTO `test`.`shipment`
(`shipmentnumber`,
`order_number`,
`order_line`,
`quantity`)
VALUES
("shipmentnumber1",
"order1",
"orderline11",
2
);

insert into test.order values("order2","orderline21","ordertype2");
insert into test.order values("order1","orderline11","ordertype1");
insert into test.order values("order2","orderline21","ordertype3");


SELECT * FROM test.`order`;


# order_number, order_line, prder_type
order2, orderline21, ordertype2
order1, orderline11, ordertype1
order2, orderline21, ordertype3


'

select * from  `test`.`shipment`\

# shipmentnumber, order_number, order_line, quantity
shipmentnumber1, order1, orderline11, 1
shipmentnumber1, order2, orderline21, 1




select * from `test`.`shipment`  as shpmnt
 left join  `test`.`order` as ord
 on
shpmnt.order_number=ord.order_number
AND
shpmnt.order_line=ord.order_line

# shipmentnumber, order_number, order_line, quantity, order_number, order_line, prder_type
'shipmentnumber1', 'order2', 'orderline21', '1', 'order2', 'orderline21', 'ordertype2'
'shipmentnumber1', 'order1', 'orderline11', '1', 'order1', 'orderline11', 'ordertype1'
'shipmentnumber1', 'order2', 'orderline21', '1', 'order2', 'orderline21', 'ordertype3'

Observe since right table is having more records we get records more than left table
============

left table also having duplicates

select * from  `test`.`shipment`

Have added duplicate record in the left table

# shipmentnumber, order_number, order_line, quantity
shipmentnumber1, order1, orderline11, 1
shipmentnumber1, order2, orderline21, 1
shipmentnumber1, order1, orderline11, 2


SELECT * FROM test.`order`;

# order_number, order_line, prder_type
order2, orderline21, ordertype2
order1, orderline11, ordertype1
order2, orderline21, ordertype3

# shipmentnumber, order_number, order_line, quantity, order_number, order_line, prder_type
shipmentnumber1, order1, orderline11, 1, order1, orderline11, ordertype1
shipmentnumber1, order1, orderline11, 2, order1, orderline11, ordertype1
shipmentnumber1, order2, orderline21, 1, order2, orderline21, ordertype2
shipmentnumber1, order2, orderline21, 1, order2, orderline21, ordertype3

all the left table records will be retained, and from the right table cross product is happening

Adding column in dataframe with user defined value if the column is null

Add another column by checking the previous column value


scala> val df1 = Seq(
     |            (
     |              "track1",
     |              "2019-11-18T16:11:28",
     |              "linekey1",
     |              "item1",
     |              "123"
     |            ),
     |            (
     |              "track2",
     |              "2019-11-18T16:11:28",
     |              "linekey2",
     |              "item2",
     |              null
     |            )).toDF(
     |            "number",
     |            "source_modified_timestamp",
     |            "line_key",
     |            "item_no",
     |            "tracking_number"
     |          )
df1: org.apache.spark.sql.DataFrame = [number: string, source_modified_timestamp: string ... 3 more fields]

scala> df1.show
+------+-------------------------+--------+-------+---------------+
|number|source_modified_timestamp|line_key|item_no|tracking_number|
+------+-------------------------+--------+-------+---------------+
|track1|      2019-11-18T16:11:28|linekey1|  item1|            123|
|track2|      2019-11-18T16:11:28|linekey2|  item2|           null|
+------+-------------------------+--------+-------+---------------+


scala>     df1.withColumn("tracking_number_rm", when($"tracking_number".isNull, lit("-999999")).otherwise(lit($"tracking_number"))).show
+------+-------------------------+--------+-------+---------------+------------------+
|number|source_modified_timestamp|line_key|item_no|tracking_number|tracking_number_rm|
+------+-------------------------+--------+-------+---------------+------------------+
|track1|      2019-11-18T16:11:28|linekey1|  item1|            123|               123|
|track2|      2019-11-18T16:11:28|linekey2|  item2|           null|           -999999|
+------+-------------------------+--------+-------+---------------+------------------+


scala>         df1.withColumn("tracking_number_rm", when($"tracking_number".isNull, "-1111").otherwise($"tracking_number")).show
+------+-------------------------+--------+-------+---------------+------------------+
|number|source_modified_timestamp|line_key|item_no|tracking_number|tracking_number_rm|
+------+-------------------------+--------+-------+---------------+------------------+
|track1|      2019-11-18T16:11:28|linekey1|  item1|            123|               123|
|track2|      2019-11-18T16:11:28|linekey2|  item2|           null|             -1111|
+------+-------------------------+--------+-------+---------------+------------------+

Group by considers null as seperate row

val simpleData = Seq(("James","Sales","NY",90000,34,10000),
    ("Michael","Sales","NY",86000,56,20000),
    ("Robert","Sales","CA",81000,30,23000),
    ("Maria","Finance","CA",90000,24,23000),
    ("Raman","Finance","CA",99000,40,24000),
    ("Scott","Finance","NY",83000,36,19000),
    ("Jen","Finance","NY",79000,53,15000),
    ("Jeff","Marketing","CA",80000,25,18000),
    ("Kumar","Marketing","NY",91000,50,21000)
  )
  val df = simpleData.toDF("employee_name","department","state","salary","age","bonus")
  df.show()

+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|        James|     Sales|   NY| 90000| 34|10000|
|      Michael|     Sales|   NY| 86000| 56|20000|
|       Robert|     Sales|   CA| 81000| 30|23000|
|        Maria|   Finance|   CA| 90000| 24|23000|
|        Raman|   Finance|   CA| 99000| 40|24000|
|        Scott|   Finance|   NY| 83000| 36|19000|
|          Jen|   Finance|   NY| 79000| 53|15000|
|         Jeff| Marketing|   CA| 80000| 25|18000|
|        Kumar| Marketing|   NY| 91000| 50|21000|
+-------------+----------+-----+------+---+-----+



    //using agg function
  df.groupBy("department")
    .agg(
      sum("salary").as("sum_salary"),
      avg("salary").as("avg_salary"),
      sum("bonus").as("sum_bonus"),
      max("bonus").as("max_bonus"))
    .show(false)

+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary       |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales     |257000    |85666.66666666667|53000    |23000    |
|Finance   |351000    |87750.0          |81000    |24000    |
|Marketing |171000    |85500.0          |39000    |21000    |
+----------+----------+-----------------+---------+---------+


=====

val simpleData = Seq(("James","Sales","NY",90000,34,10000),
    ("Michael","Sales","NY",86000,56,20000),
    ("Robert","Sales","CA",81000,30,23000),
    ("Maria","Finance","CA",90000,24,23000),
    ("Raman","Finance","CA",99000,40,24000),
    ("Scott","Finance","NY",83000,36,19000),
    ("Jen","Finance","NY",79000,53,15000),
    ("Jeff","Marketing","CA",80000,25,18000),
    ("Kumar",null,"NY",91000,50,21000)
  )
  val df = simpleData.toDF("employee_name","department","state","salary","age","bonus")
  df.show()


  +-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|        James|     Sales|   NY| 90000| 34|10000|
|      Michael|     Sales|   NY| 86000| 56|20000|
|       Robert|     Sales|   CA| 81000| 30|23000|
|        Maria|   Finance|   CA| 90000| 24|23000|
|        Raman|   Finance|   CA| 99000| 40|24000|
|        Scott|   Finance|   NY| 83000| 36|19000|
|          Jen|   Finance|   NY| 79000| 53|15000|
|         Jeff| Marketing|   CA| 80000| 25|18000|
|        Kumar|      null|   NY| 91000| 50|21000|
+-------------+----------+-----+------+---+-----+

Observe null is treated as seperate row

+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary       |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales     |257000    |85666.66666666667|53000    |23000    |
|null      |91000     |91000.0          |21000    |21000    |
|Finance   |351000    |87750.0          |81000    |24000    |
|Marketing |80000     |80000.0          |18000    |18000    |
+----------+----------+-----------------+---------+---------+

Loading Hive config in spark shell

-bash-4.2$ spark-shell --conf spark.sql.sources.partitionOverwriteMode=dynamic \
>        --conf spark.hadoop.hive.exec.dynamic.partition.mode=nonstrict \
>        --conf spark.sql.orc.impl=native \
>        --conf spark.sql.orc.enableVectorizedReader=true \
>        --conf spark.hadoop.hive.exec.max.dynamic.partitions=2550



Observe spark.hadoop.* is placed for all the hadoop configurations


If we pass  only --conf hive.exec.max.dynamic.partitions=2550 then this property will be ignored by spark-shell in 2.3.1 version


We can also pass memory config as below

spark-shell --driver-memory 10G --executor-memory 15G --executor-cores 8