import numpy as np
import matplotlib.pyplot as plt
# Consider method "load_house_data" load the dataset
X_train, y_train = load_house_data()
X_features = ['size(sqft)','bedrooms','floors','age']
fig,ax=plt.subplots(1, 4, figsize=(12, 3), sharey=True)
for i in range(len(ax)):
ax[i].scatter(X_train[:,i],y_train)
ax[i].set_xlabel(X_features[i])
ax[0].set_ylabel("Price (1000's)")
plt.show()
Sample code to generate rows in oracle using recursive with clause
with ab(cnt) as
(select 2 as cnt from dual
union all
select cnt+2 from ab where cnt<5
)
select * from ab;
I'm trying to use rank function in sql server with repeating value in the "partition by" clause, but I can't get exactly what I need. This is my query :
select jobID, runStatus,
rank() over (partition by runStatus order by jobID ) as rank
from table
and I get:
jobID runStatus rank 10 s 9 9 s 8 8 s 7 7 s 6 6 s 5 5 s 4 4 f 1 3 s 3 2 s 2 1 s 1
but what I really need is:
jobID runStatus rank desire 10 s 6 9 s 5 8 s 4 7 s 3 6 s 2 5 s 1 4 f 1 3 s 3 2 s 2 1 s 1
meaning that the rank will be initialized every change in runStatus column.
Below is the solution:
You need to define the groups. This is called a "gap-and-islands" problem. And one solution is the difference of row numbers. For your data, this looks like:
select jobID, runStatus,
row_number() over (partition by runStatus, seqnum - seqnum_rs
order by jobID
) as rank
from (select t.*,
row_number() over (order by jobId) as seqnum,
row_number() over (partition by runStatus order by jobId) as seqnum_rs
from t
) t;import sys
try:
with open('journal.txt', 'a') as f:
f.write('input')
except IOError as exc: # Python 2. For Python 3 use OSError
tb = sys.exc_info()[-1]
lineno = tb.tb_lineno
filename = tb.tb_frame.f_code.co_filename
print('{} at {} line {}.'.format(exc.strerror, filename, lineno))
sys.exit(exc.errno)
--Changing the table type from external table to managed table.
alter table c_sample_tbl set tblproperties('EXTERNAL'='FALSE');
--Dropping the hive partition along with its cache.
alter table c_sample_tbl drop if exists partition (bus_dt='2019-09-30') purge;
--Renaming the hive partition.
ALTER TABLE c_sample_tbl PARTITION (bus_dt='2019-09-30_1') RENAME TO PARTITION (bus_dt='2019-09-30');
--Changing the table type from managed table to external table.
alter table c_sample_tbl set tblproperties('EXTERNAL'='TRUE');
--Inserting sample data /static data to hive table for a static partition.
insert into c_curr_rates PARTITION (bus_dt = '2019-06-30') values ('INR','SGD',0.01886,'2019-12-02');
Linux command to run in background.
command: nohup sh run.sh &
Linux command to run multiple commands in sequence.
command: nohup sh run.sh 2021-01-05 && sh run.sh 2021-02-08 && sh run.sh 2021-03-09 &
Finding the back ground running process in linux and killing it.
command: ps -ef | grep sh run.sh
get the pid from above command output and use it to kill it.
kill -9 92646
**************Py Spark Multiple Sessions********************************
#Initialize first pyspark session as below
import pyspark
spark1=pyspark.sql.SparkSession.builder.master("local").appName("Single_app").getOrCreate()
#Tempory view created in one session cannot be accessed in another session, below is an example
spark2=spark1.newSession()
spark1.sql("select 1,2").createOrReplaceTempView("spark1_view")
spark1.catalog.listTables()
spark2.catalog.listTables()
#To access or have a common temp space, create a global temp view
spark1.sql("select 1,2").createGlobalTempView("spark1_global")
spark2.sql("select * from global_temp.spark1_global").show()
"""
Another thing , if we have different sessions sharing same context we can have different set of configurations among
those sessions as below
"""
spark1.conf.set("spark.sql.shuffle.partitions","100")
spark2.conf.set("spark.sql.shuffle.partitions","200")
#display the respective conf values
spark1.conf.get("spark.sql.shuffle.partitions")
spark2.conf.get("spark.sql.shuffle.partitions")
#stopping one session stops SparkContext and all associated sessions
spark1.stop()
************************Below is scala version***********************
he two most common uses cases are:
Keeping sessions with minor differences in configuration.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.2.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_141)
Type in expressions to have them evaluated.
Type :help for more information.
scala> spark.range(100).groupBy("id").count.rdd.getNumPartitions
res0: Int = 200
scala>
scala> val newSpark = spark.newSession
newSpark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@618a9cb7
scala> newSpark.conf.set("spark.sql.shuffle.partitions", 99)
scala> newSpark.range(100).groupBy("id").count.rdd.getNumPartitions
res2: Int = 99
scala> spark.range(100).groupBy("id").count.rdd.getNumPartitions // No effect on initial session
res3: Int = 200
Separating temporary namespaces:
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.2.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_141)
Type in expressions to have them evaluated.
Type :help for more information.
scala> spark.range(1).createTempView("foo")
scala>
scala> spark.catalog.tableExists("foo")
res1: Boolean = true
scala>
scala> val newSpark = spark.newSession
newSpark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@73418044
scala> newSpark.catalog.tableExists("foo")
res2: Boolean = false
scala> newSpark.range(100).createTempView("foo") // No exception
scala> spark.table("foo").count // No effect on inital session
res4: Long = 1 #Consider following data as existing data in target table.
dat= [[1, 50, "2019-02-01", "2019-02-02", 0],
[1, 75, "2019-02-02", None,1],
[2, 200, "2019-02-01", "2019-02-01", 0],
[2, 60, "2019-02-01", "2019-02-01", 0],
[2, 500, "2019-02-01", None, 1],
[3, 175, "2019-02-01", None, 1],
[4, 50, "2019-02-02", "2019-02-02", 0],
[4, 300, "2019-02-02", None, 1],
[5, 500, "2019-02-02", None, 1]]
header=["pk", "amount", "StartDate", "endDate", "active"]
df_existing_data=spark.createDataFrame(data=dat,schema=header)
#Create a table with SCD Type 2 and associated structure.
df_existing_data.write.format("parquet").mode("overwrite").partitionBy("active").saveAsTable("default.tab_amount")
df_existing_data.show()
#Extracting current Data from target table/scd type 2 table.
df_active_data=spark.sql("select * from default.tab_amount where active=1")
df_active_data.createOrReplaceTempView("ExisingActiveData")
df_active_data.show()
#Consider following as latest or new data arrived from source system.
dat= [
[1, 75],
[2, 500],
[3, 200],
[4, 350],
[5, 500],
[6, 800],
[7, 1500]]
header=["pk", "amount"]
df_newdata=spark.createDataFrame(data=dat,schema=header)
df_newdata.createOrReplaceTempView("NewData")
df_newdata.show()
#Extracting new records to insert with start date as today and end date as NULL.
df_NewRecords=spark.sql("select ND.*,current_Date() as startDate,NULL as endDate,1 as active from NewData ND left anti join ExisingActiveData EAD on EAD.pk=ND.pk ")
df_NewRecords.show()
#Extracting updated active records from source table with start date as today and end date as NULL.
df_UpdatedActive=spark.sql("select ND.*,current_date() as startDate,cast(NULL as string) as endDate,1 as active from NewData ND inner join ExisingActiveData EAD on EAD.pk=ND.pk where EAD.amount<>ND.amount")
df_UpdatedActive.createOrReplaceTempView("UpdatedActive")
df_UpdatedActive.show()
#Extracting existing active records where the values are not changed in target vs source comparison.
df_ExistingActive=spark.sql("select EAD.* from ExisingActiveData EAD left anti join UpdatedActive UA on EAD.pk=UA.pk")
df_ExistingActive.show()
#Extracting old updated records for updating its end date to yesterday.
df_UpdatedInactive=spark.sql("select EAD.pk,EAD.amount,startDate,date_sub(current_date(),1) as endDate,0 as active from ExisingActiveData EAD inner join NewData ND on EAD.pk=ND.pk where EAD.amount<>ND.amount")
df_UpdatedInactive.show()
#Configuring for auto insert and dynamic partitioning.
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
#Consolidated active records.
df_TotalActive=df_NewRecords.unionAll(df_UpdatedActive).unionAll(df_ExistingActive)
#Appending the updated records with end date as yesterday to active=0 partition
df_UpdatedInactive.write.mode("append").insertInto("default.tab_amount")
#Overriding the consolidated active records to active=1 partition
df_TotalActive.write.mode("overwrite").insertInto("default.tab_amount")
#Below is the over all data in scd type 2 dimension after data load.
spark.sql("select * from default.tab_amount order by pk,active asc").show()
The main components of the Hive are as follows:
While Hadoop/hive can process nearly any amount of data, but optimizations can lead to big savings, proportional to the amount of data, in terms of processing time and cost. There are a whole lot of optimizations that can be applied in the hive. Let us look into the optimization techniques we are going to cover:
One way to control the size of files when inserting into a table using Hive, is to set the below parameters:
set hive.merge.tezfiles=true;
set hive.merge.mapfiles=true;
set hive.merge.mapredfiles=true;
set hive.merge.size.per.task=128000000;
set hive.merge.smallfiles.avgsize=128000000;
This will work for both M/R and Tez engine and will ensure that all files created are at or below 128 MB in size (you can alter that size number according to your use case. Additional reading here: https://community.cloudera.com/t5/Community-Articles/ORC-Creation-Best-Practices/ta-p/248963).
The easiest way to merge the files of the table is to remake it, while having ran the above hive commands at runtime:
CREATE TABLE new_table LIKE old_table;
INSERT INTO new_table select * from old_table;Create a table with array, struct complex data types as below.
CREATE TABLE student_details(
id_key string,
name string,
subjects array<string>,
address struct<city:string,State:string>
);
Insert sample data as below.
INSERT INTO student_details with below command.
select
'AA87U',
'BRYAN',
array('ENG','CAL_1','CAL_2','HST','MUS'),
named_struct('city','Tampa','State','FL');
display the sample data.
select * from student_details;
