source: gs3-extensions/maori-lang-detection/hdfs-instructions/scripts/export_maori_index_csv.sh@ 33524

#!/bin/bash

# convert URL index with Spark on Yarn

# Table format configuration
FORMAT=${FORMAT:-parquet} # parquet, orc
NESTED="$NESTED"          # "" (empty) or --useNestedSchema
COMPRS=${COMPRS:-gzip}    # gzip, snappy, lzo, none
PARTITION_BY="crawl,subset"

# Input spec (URL index files to convert)
#DATA=s3a://commoncrawl/cc-index/collections/CC-MAIN-2017-51/indexes/cdx-*.gz
#DATA="$1"
# Output directory
#OUTPUTDIR=hdfs:///user/ubuntu/cc-index-table/
CRAWL_ID=$1
OUTPUTDIR="hdfs:///user/vagrant/${CRAWL_ID}/cc-mri-csv"
OUTPUT_PARENTDIR=hdfs:///user/vagrant/${CRAWL_ID}
     # or just OUTPUT_PARENTDIR=/user/vagrant/${CRAWL_ID}, since /user/vagrant is on hdfs:

# https://stackoverflow.com/questions/26513861/checking-if-directory-in-hdfs-already-exists-or-not
hdfs dfs -test -d $OUTPUT_PARENTDIR
if [ $? == 0 ]; then
    echo "Directory $OUTPUT_PARENTDIR already exists."        
else
    echo "Creating directory $OUTPUT_PARENTDIR..."    
    hdfs dfs -mkdir $OUTPUT_PARENTDIR
fi


# Spark configuration
SPARK_HOME="$SPARK_HOME"
# EXECUTOR_MEM=44g
# EXECUTOR_CORES=12
# NUM_EXECUTORS=4
# DRIVER_MEM=4g

EXECUTOR_MEM=3g
EXECUTOR_CORES=2
NUM_EXECUTORS=2
DRIVER_MEM=3g


SPARK_ON_YARN="--master yarn"
SPARK_EXTRA_OPTS=""

# source specific configuration file
## test -e $(dirname $0)/convert_url_index_conf.sh && . $(dirname $0)/convert_url_index_conf.sh


_APPJAR=$PWD/target/cc-spark-0.2-SNAPSHOT-jar-with-dependencies.jar

##export LIBJARS=/home/vagrant/lib/hadoop-aws-2.7.6.jar
#export LIBJARS=/home/vagrant/lib/*
#export HADOOP_CLASSPATH=`echo ${LIBJARS} | sed s/,/:/g`

set -e
set -x


#   --conf spark.hadoop.fs.s3a.access.key=ACCESSKEY \
#   --conf spark.hadoop.fs.s3a.secret.key=SECRETKEY \
#    --conf spark.hadoop.fs.s3a.access.key=AKIA2EVQBWSTBJ2M4BLM \
#    --conf spark.hadoop.fs.s3a.secret.key=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 \


#    $SPARK_ON_YARN \
# /home/vagrant/.m2/repository/org/apache/hadoop/hadoop-aws/2.7.6/hadoop-aws-2.7.6.jar
#    --jars file:/home/vagrant/aws-java-sdk-1.7.4.jar,file:/home/vagrant/lib/hadoop-aws-2.7.6.jar \
#    --driver-class-path=/home/vagrant/lib/aws-java-sdk-1.7.4.jar:/home/vagrant/lib/hadoop-aws-2.7.6.jar \

# https://www.patricia-anong.com/blog/2017/11/1/extend-vmdk-on-virtualbox

$SPARK_HOME/bin/spark-submit \
    --jars file:/home/vagrant/.m2/repository/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar,file:/home/vagrant/.m2/repository/org/apache/hadoop/hadoop-aws/2.7.6/hadoop-aws-2.7.6.jar \
    --driver-class-path=/home/vagrant/.m2/repository/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar:/home/vagrant/.m2/repository/org/apache/hadoop/hadoop-aws/2.7.6/hadoop-aws-2.7.6.jar \
    --conf spark.hadoop.parquet.enable.dictionary=true \
    --conf spark.hadoop.parquet.enable.summary-metadata=false \
    --conf spark.sql.hive.metastorePartitionPruning=true \
    --conf spark.sql.parquet.filterPushdown=true \
    --conf spark.sql.parquet.mergeSchema=true \
    --class org.commoncrawl.spark.examples.CCIndexExport $_APPJAR \
    --query "SELECT url, warc_filename, warc_record_offset, warc_record_length   
            FROM ccindex                                                        
            WHERE crawl = '${CRAWL_ID}' AND subset = 'warc' AND content_languages = 'mri'" \
    --outputFormat csv \
    --numOutputPartitions 10 \
    --outputCompression gzip \
    s3a://commoncrawl/cc-index/table/cc-main/warc/ \
    $OUTPUTDIR