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

#!/bin/bash

# This script is BASED ON the cc-index-table github project's convert_url_index script at
# https://github.com/commoncrawl/cc-index-table/blob/master/src/script/convert_url_index.sh
# That script is described as
# "A Spark job converts the Common Crawl URL index files (a sharded gzipped index in CDXJ format)
# into a table in Parquet or ORC format." (https://github.com/commoncrawl/cc-index-table)."
# If you want to run that script, then modify its variables to have the following values before
# running it, in order for it to work on our machine for doing analytics:
# EXECUTOR_MEM=3g
# EXECUTOR_CORES=2
# NUM_EXECUTORS=2
# DRIVER_MEM=3g
# Since that script was copied here, as a result, a lot of such variables (like executor
# and memory related) are unused here, as they were just copied directly across. Such unused
# variables can probably be removed from this file.

# This script was modified to do the following:
# SQL query CommonCrawl's distributed cc-index table on Amazon S3 for the parameterised crawl timestamp
# and get all those records for which the primary language in the content_languages field is MRI for Maori.
# Only the WARC related fields (url, filename, offset and length fields) of each record are requested.
# The matching records' fields are then constructed into a distributed csv file on the local hdfs system
# A second phase then requests the warc files at those offsets and downloads them onto the local hdfs.
# We still get zipped WARC files, but they only contain the pages of that crawl where the primary language
# was identified as MRI.
# A third phase converts those WARC files into WET (and WAT) files and copies these zipped files onto the
# mounted shared space on vagrant.


#---------------------------- START UNUSED VARIABLES---------------------------#

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


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

#--- Dr Bainbridge modified the above variables in the original script, convert_url_index.sh,
# as follows in order to get that spark job to run. Not used in this script. ---#
EXECUTOR_MEM=3g
EXECUTOR_CORES=2
NUM_EXECUTORS=2
DRIVER_MEM=3g

#--- VARIABLES PROBABLY ALSO NOT OF USE IN THIS SCRIPT ---#
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

#---------------------------- END UNUSED VARIABLES---------------------------#


# The crawl timestamp, of the form CC-MAIN-2019-26
# Obtain from http://index.commoncrawl.org/
CRAWL_ID=$1
if [ "x$CRAWL_ID" == "x" ]; then
    echo "No crawl timestamp provided. Should be of the form CC-MAIN-YYYY-COUNT."
    echo "e.g. CC-MAIN-2019-26. Choose a crawl timestamp from http://index.commoncrawl.org/"
    exit
fi
# Output directory
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
if $(hdfs dfs -test -d "$OUTPUT_PARENTDIR"); then
    echo "Directory $OUTPUT_PARENTDIR already exists."        
else
    echo "Creating directory $OUTPUT_PARENTDIR..."    
    hdfs dfs -mkdir $OUTPUT_PARENTDIR
fi


_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

# PHASE 1: querying this crawl's massive index with an SQL query that requests just the references to warc files
# for those crawled web pages where the content_languages field's primary language is MRI (3 letter code for Maori)
# The output is a distributed .csv file which will be stored in a "cc-mri-csv" subfolder of the $OUTPUT_PARENTDIR.

#OUTPUTDIR="hdfs:///user/vagrant/${CRAWL_ID}/cc-mri-csv"
OUTPUTDIR="${OUTPUT_PARENTDIR}/cc-mri-csv"

#   --conf spark.hadoop.fs.s3a.access.key=ACCESSKEY \
#   --conf spark.hadoop.fs.s3a.secret.key=SECRETKEY \


#    $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


# The above generates ZIPPED part*.csv files in $OUTPUTDIR (folder cc-mri-csv).
# First create a folder and unzip into it:
#hdfs dfs -test -d $OUTPUT_PARENTDIR/cc-mri-unzipped-csv
#if [ $? == 0 ]; then
if $(hdfs dfs -test -d "$OUTPUT_PARENTDIR/cc-mri-unzipped-csv"); then
    echo "Directory cc-mri-unzipped-csv already exists for crawl ${CRAWL_ID}."
    echo "Assuming cc-mri.csv also exists inside $OUTPUT_PARENTDIR"
else
    echo "Creating directory $OUTPUT_PARENTDIR/cc-mri-unzipped-csv..."
    hdfs dfs -mkdir $OUTPUT_PARENTDIR/cc-mri-unzipped-csv

    echo "Unzipping ${OUTPUTDIR}/part files into $OUTPUT_PARENTDIR/cc-mri-unzipped-csv/cc-mri.csv"
    hdfs dfs -cat $OUTPUTDIR/part* | gzip -d | hdfs dfs -put - $OUTPUT_PARENTDIR/cc-mri-unzipped-csv/cc-mri.csv
fi



# PHASE 2, which uses the index of MRI warc URLs and offsets,
# stored in the now unzipped .csv file,
# to get all the WARC records it specifies at the specified warc offsets.
# This will get us all the MRI pages of the commoncrawl for CRAWL_ID

# Change OUTPUTDIR
OUTPUTDIR="hdfs:///user/vagrant/${CRAWL_ID}/warc"


#   --conf spark.hadoop.fs.s3a.access.key=ACCESSKEY \
#   --conf spark.hadoop.fs.s3a.secret.key=SECRETKEY \


#    $SPARK_ON_YARN \
$SPARK_HOME/bin/spark-submit \
   --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
   --class org.commoncrawl.spark.examples.CCIndexWarcExport $_APPJAR \
   --csv $OUTPUT_PARENTDIR/cc-mri-unzipped-csv/cc-mri.csv \
   --numOutputPartitions 10 \
   --numRecordsPerWarcFile 5000 \
   --warcPrefix MAORI-${CRAWL_ID} \
   s3a://commoncrawl/cc-index/table/cc-main/warc/ \
   $OUTPUTDIR


# PHASE 3: convert warc files to wet files and copy the wet files into the mounted shared area

#hdfs dfs -test -f $OUTPUTDIR/_SUCCESS
#if [ $? == 0 ]; then
if $(hdfs dfs -test -f "$OUTPUTDIR/_SUCCESS"); then
    # ia-hadoop-tools converts warc files into wet (and wat) files but expects a particular folder structure
    # Create the expected folder structure: a "wet" and a "wat" folder should exist
    # at the same level as the "warc" folder ($OUTPUTDIR) that contains the warc gz files.
    
    echo "Creating wet (text) and wat (metadata) directories to put converted warc files into..."
    hdfs dfs -mkdir $OUTPUT_PARENTDIR/wet
    hdfs dfs -mkdir $OUTPUT_PARENTDIR/wat
    # move the _SUCCESS file from /warc to its parent folder
    hdfs dfs -mv $OUTPUTDIR/_SUCCESS $OUTPUT_PARENTDIR/.

    # Move from the "/home/vagrant/cc-index-table" github project to the "ia-hadoop-tools" github project:
    cd /home/vagrant/ia-hadoop-tools
    # PWD is now ia-hadoop-tools folder
    # To run the following, guava.jar should be on the hadoop classpath
    # This can be achieved with:
    # vagrant@node1:~/ia-hadoop-tools$ sudo cp /usr/share/java/guava.jar /usr/local/hadoop/share/hadoop/common/.
    $HADOOP_MAPRED_HOME/bin/hadoop jar $PWD/target/ia-hadoop-tools-jar-with-dependencies.jar WEATGenerator -strictMode -skipExisting batch-id-xyz $OUTPUTDIR/*.warc.gz

    # Now copy the zipped wet-files into the mounted folder, so we can scp the files from there to where WETProcessor.java is
    echo "Copying and tarring up the wet files"
    mkdir /vagrant/${CRAWL_ID}-wet-files
    hdfs dfs -get $OUTPUT_PARENTDIR/wet/*.warc.wet.gz /vagrant/${CRAWL_ID}-wet-files/.
    cd /vagrant
    #tar -cvf ${CRAWL_ID}.tar /vagrant/${CRAWL_ID}-wet-files
    #rm -rf /vagrant/${CRAWL_ID}-wet-files
fi