ノートブックの作業

%pyspark
from pyspark.sql import SparkSession
sparkSession = SparkSession.builder.appName("SparkSessionName").getOrCreate()

%pyspark
df_load = sparkSession.read.csv('/hdfs/full/path/to/folder') 
df_load.show()

%pyspark
data = [('One', 1), ('Two', 2), ('Three', 3)]
# Create DataFrame
df = sparkSession.createDataFrame(data)
# Write into HDFS 
df.write.mode("overwrite").csv("/hdfs/full/path/to/folder")
# /hdfs/full/path/to/folder is created when dfs.write is executed

• append: このDataFrameの内容を既存のデータに追加します。

• overwrite: 既存のデータを上書きします。

• ignore: データがすでに存在する場合、この操作は無視されます。

• errorまたはerrorifexists: データがすでに存在する場合は、例外をスローします。

%pyspark
from pyspark.sql import SparkSession
sparkSession = SparkSession.builder.appName("pyspark-write").getOrCreate()
# Create data
data = [('One', 1), ('Two', 2), ('Three', 3), ('Four', 4), ('Five', 5)]
df = sparkSession.createDataFrame(data)
# Write into HDFS 
df.write.mode("overwrite").csv("/tmp/testpysparkcsv")

%pyspark
from pyspark.sql import SparkSession
sparkSession = SparkSession.builder.appName("pyspark-read").getOrCreate()
# Read from HDFS
df_load = sparkSession.read.csv('/tmp/testpysparkcsv')

df_load.show()

%pyspark
# Run the SQL instruction in quiet mode
spark.sql("SQL instruction")
# Show some output
spark.sql("SQL instruction").show()

%pyspark
spark.sql ("DROP TABLE IF EXISTS hive_table")
spark.sql("CREATE TABLE IF NOT EXISTS hive_table (number int, Ordinal_Number string, Cardinal_Number string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' ")
spark.sql("load data inpath '/tmp/pysparktestfile.csv' into table pyspark_numbers_from_file")
spark.sql("insert into table pyspark_numbers_from_file2 select * from pyspark_numbers_from_file")
spark.sql("CREATE TABLE IF NOT EXISTS pyspark_numbers (number int, Ordinal_Number string, Cardinal_Number string) USING com.databricks.spark.csv OPTIONS \
(path \"/full/path/hdfs/dataframe/folder\", header \"false\")")

%pyspark
spark.sql("show databases").show()
spark.sql("show tables").show()
spark.sql("describe hive_table ").show()
spark.sql("select * hive_Table limit 5").show()

%pyspark
spark.sql("CREATE TABLE IF NOT EXISTS pyspark_test_table ( eid int, name String,  salary String, destination String)")
spark.sql("describe pyspark_test_table ").show()

%pyspark
spark.sql ("DROP TABLE IF EXISTS pyspark_test_table")

%pyspark
from pyspark.sql import SparkSession
import re

single_digit = ["","one", "two", "three" , "four" , "five" , "six" , "seven" , "eight" , "nine"] 
ten_to_nineteen  =["ten","eleven","twelve","thirteen" ,"fourteen","fifteen","sixteen","seventeen","eighteen","nineteen"]
two_digits  =["","","twenty","thirty","fourty","fithy","sixty","seventy","eighty","ninety"]
ten_power = ["hundred","thousand"]
data = []

def cardinalNum(number):
    if len (str(number)) == 1 :
        return single_digit[number]
    elif len (str(number)) == 2 :
        if str(number)[0] == "1" :
            return ten_to_nineteen[int(str(number)[1])]
        else:
            return two_digits[int(str(number)[0])] + " " + single_digit[int(str(number)[1])]
    elif len (str(number)) == 3 :
        return   single_digit[int(str(number)[0])] + " " + ten_power[0] + " " +  cardinalNum(int(str(number)[1:3]))
    elif len (str(number)) == 4 :
        return cardinalNum(int(str(number)[0:1])) + " " + ten_power[1] + " " + cardinalNum(int(str(number)[1:4]))
    else:
        return str(number)

def ordinalNum(number):
    if re.match(".*1[0-9]$", str(number)):
        return(str(number) + 'th')
    elif re.match(".*[04-9]$", str(number)):
        return(str(number) + 'th')
    elif re.match(".*1$", str(number)):
        return(str(number) + 'st')
    elif re.match(".*2$", str(number)):
        return(str(number) + 'nd')
    elif re.match(".*3$", str(number)):
        return(str(number) + 'rd')
    else:
        return(str(number))

sparkSession = SparkSession.builder.appName("pyspark-write").getOrCreate()

# Create data
for number in range(1, 9001):
    tmpdata=[ (number) ,( ordinalNum(number)) ,(cardinalNum(number))]
    data.append(tmpdata) 
    #print ( str(number) + ',' + ordinalNum(number) + ',' + cardinalNum(number))

df = sparkSession.createDataFrame(data)

# Write into HDFS
df.write.mode("overwrite").csv("/tmp/testpyspark")

%pyspark
spark.sql("CREATE TABLE IF NOT EXISTS pyspark_numbers (number int, Ordinal_Number string, Cardinal_Number string) USING com.databricks.spark.csv OPTIONS \
(path \"/tmp/testpyspark\", header \"false\")")
spark.sql ("describe pyspark_numbers").show()
spark.sql ("select * from pyspark_numbers").show()
spark.sql("select count(*) from pyspark_numbers").show()

%pyspark
spark.sql("CREATE TABLE IF NOT EXISTS pyspark_numbers_from_file (number int, Ordinal_Number string, Cardinal_Number string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' ")
spark.sql("load data inpath '/tmp/pysparktestfile.csv' into table pyspark_numbers_from_file")

# Command listing files
hdfs dfs  -ls /tmp/testpyspark/

# Output listing files
Found 2 items
-rw-r--r--   3 datastudio supergroup          0 2020-08-21 16:50 /tmp/testpyspark/_SUCCESS
-rw-r--r--  
 3 datastudio supergroup     416250 2020-08-21 16:50 
/tmp/testpyspark/part-00000-5bf7c92a-e2ad-4f79-802e-c84c0c3b4cc0-c000.csv

# Command copying file
hdfs dfs -cp -p /tmp/testpyspark/part-00000-5bf7c92a-e2ad-4f79-802e-c84c0c3b4cc0-c000.csv /tmp/pysparktestfile.csv

%pyspark
spark.sql("CREATE TABLE IF NOT EXISTS pyspark_numbers_new (number int, Ordinal_Number string, Cardinal_Number string) ")
spark.sql("insert into table pyspark_numbers_new  select * from pyspark_numbers_from_file")
spark.sql("select * from  pyspark_numbers_new").show()
spark.sql("select count(*) from pyspark_numbers_new").show()
spark.sql("select count(*) from pyspark_numbers_from_file").show()