<jupyter_start><jupyter_text>This notebook is to help automatically import csv file to hive<jupyter_code>%%sh
mkdir -p /v3io/${V3IO_HOME}/examples
mkdir -p /v3io/${V3IO_HOME}/examples/csvc

# Download a sample stocks file from Iguazio demo bucket in S3
curl -L "https://s3.wasabisys.com/iguazio/data/stocks/2018-03-26_BINS_XETR08.csv" > /v3io/${V3IO_HOME}/examples/demo.csv
curl -L "https://s3.wasabisys.com/iguazio/data/stocks/2018-03-26_BINS_XETR08.csv" > /v3io/${V3IO_HOME}/examples/csvs/example1.csv
curl -L "https://s3.wasabisys.com/iguazio/data/stocks/2018-03-26_BINS_XETR08.csv" > /v3io/${V3IO_HOME}/examples/csvs/example2.csv<jupyter_output>% Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  861k  100  861k    0     0  7425k      0 --:--:-- --:--:-- --:--:-- 7425k
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  861k  100  861k    0     0  7362k      0 --:--:-- --:--:-- --:--:-- 7362k
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  861k  100  861k    0     0  7760k      0 --:--:-- --:--:-- --:--:-- 7760k<jupyter_text>Below is import of all needed dependencies. And in this sell you should pass path where parquet files located.<jupyter_code>import os
HOME = os.getenv('V3IO_HOME')
V3IO_HOME_URL=os.getenv('V3IO_HOME_URL')<jupyter_output><empty_output><jupyter_text>Here is creating of spark context with hive support.<jupyter_code>from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Import parquet schema to hive").config("hive.metastore.uris", "thrift://hive:9083").enableHiveSupport().getOrCreate()<jupyter_output><empty_output><jupyter_text>Define function below for getting sql script needed for creating table in hive using dataframe types as columns to table<jupyter_code>def getCreateTableScriptCSV(databaseName, tableName, path, df):
    cols = df.dtypes
    createScript = "CREATE EXTERNAL TABLE IF NOT EXISTS " + databaseName + "." + tableName + "("
    colArray = []
    for colName, colType in cols:
        colArray.append(colName.replace(" ", "_") + " " + colType)
    createColsScript =   ", ".join(colArray )
    
    script = createScript + createColsScript + ") ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION '" + path + "' TBLPROPERTIES('skip.header.line.count'='1') "
    print (script)
    return script
#define main function for creating table where arqument 'path' is path to parquet files 
def createTableCSV(databaseName, tableName, path): 
    df = spark.read.format("csv").option("header", "true").option("inferschema","true").load(path)
    sqlScript = getCreateTableScriptCSV(databaseName, tableName, path, df)
    spark.sql(sqlScript)<jupyter_output><empty_output><jupyter_text>One file example<jupyter_code># Set path where the csv file located.
my_csv_file_path = os.path.join(V3IO_HOME_URL, 'examples/demo.csv')
createTableCSV("default","csv_table",my_csv_file_path)<jupyter_output>CREATE EXTERNAL TABLE IF NOT EXISTS default.csv_table(ISIN string, Mnemonic string, SecurityDesc string, SecurityType string, Currency string, SecurityID int, Date timestamp, Time string, StartPrice double, MaxPrice double, MinPrice double, EndPrice double, TradedVolume int, NumberOfTrades int) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'v3io://users/admin/examples/demo.csv' TBLPROPERTIES('skip.header.line.count'='1')<jupyter_text>One folder example for spark output job<jupyter_code># Set path where parquet folder with csv files inside located.
folder_path = os.path.join(V3IO_HOME_URL, 'examples/csvs/')
createTableCSV("default","table_from_dir2",folder_path)<jupyter_output>CREATE EXTERNAL TABLE IF NOT EXISTS default.table_from_dir2(ISIN string, Mnemonic string, SecurityDesc string, SecurityType string, Currency string, SecurityID string, Date string, Time string, StartPrice string, MaxPrice string, MinPrice string, EndPrice string, TradedVolume string, NumberOfTrades int) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'v3io://users/admin/examples/csvs/' TBLPROPERTIES('skip.header.line.count'='1')<jupyter_text>Multiple files and folders exampleWrite here name of database and path to folder where all csv's files (or folders with them) located. Database should be created.In this cell code goes over all files and dirs in provided path and using them for creating table.File should be ended with .csv format and should be "," seperated.Directory (in which stored csv files) should be started with "."Name of directory or file will be name of table.<jupyter_code>databaseName = "default"
filepath = os.path.join('/v3io', HOME, 'examples/csvs')

for fileOrDir in os.listdir(filepath):
    if fileOrDir.endswith(".csv") :
        createTableCSV(databaseName, fileOrDir.split(".csv")[0], filepath.replace("/v3io/", "v3io://", 1) + "/" + fileOrDir)
    elif not fileOrDir.startswith(".") :
        createTableCSV(databaseName, fileOrDir, filepath.replace("/v3io/", "v3io://", 1) + "/" + fileOrDir + "/*")<jupyter_output>CREATE EXTERNAL TABLE IF NOT EXISTS default.demo1(id int, street string, city string, zip int, state string, beds int, baths int, sq__ft int, type string, sale_date string, price int, latitude double, longitude double) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'v3io://users/admin/examples/csvs/demo1.csv' TBLPROPERTIES('skip.header.line.count'='1') 
CREATE EXTERNAL TABLE IF NOT EXISTS default.demo2(id int, street string, city string, zip int, state string, beds int, baths int, sq__ft int, type string, sale_date string, price int, latitude double, longitude double) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'v3io://users/admin/examples/csvs/demo2.csv' TBLPROPERTIES('skip.header.line.count'='1') 
CREATE EXTERNAL TABLE IF NOT EXISTS default.example1(ISIN string, Mnemonic string, SecurityDesc string, SecurityType string, Currency string, SecurityID int, Date timestamp, Time string, StartPrice double, MaxPrice double, MinPrice double, EndPrice double, TradedVolume in[...]<jupyter_text>Test how it works<jupyter_code># test how the tables were saved
#spark.sql("drop database test CASCADE")
spark.sql("show databases").show()
spark.sql("show tables in " + databaseName).show()
# test how saving to table works
tableName = "example1"
spark.sql("select * from " + databaseName + "." + tableName)
spark.sql("drop table " + databaseName + ".example1")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11453017111978385, 'fraction_numerical': 0.022515761032722906, 'mean_word_length': 4.37772397094431, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 32, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '613218', 'license': 'Apache-2.0', 'n_tokens_mistral': 2271, 'n_tokens_neox': 2117, 'n_words': 603, 'path': 'data-ingestion-and-preparation/csv-to-hive.ipynb', 'repo_name': 'jasonnIguazio/tutorials', 'size': '6662'}