<jupyter_start><jupyter_text>Moving Data Around* When you need to process a lot of data, a big part of the execution time of your program is devoted to moving the data between storage units. * This notebook is **NOT** intended to be run on your personal computer. It is intended to show you the main steps needed when processing a large file on a multi-computer cluster. The main emphasis here is on the **Wall time** required for different operations.  Some terminology  Data Serialization.* Data in memory is usually stored in **data structures** that allow for fast manipulation. This often means that the amount of memory needed is significantly larger than the amount that would be needed to store the same data on disk.* We say that the data on disk is **serial** and the data stored in data structures is **deserialized**  AWS-EMRWe demonstrate the movement of data on "Amazon Web Services" (AWS) "Elastic Map Reduce" (EMR).Recall the slide about data organization in the video "a short history of affordable massive computing"  In the next figure we add to that slide the way it fits within AWS-EMR.  Three file systems: * **S3:** long term persistent memory. * **Head Node:** standard Unix file system. * **HDFS:** distributed file system on the workers.   Reading a CSV file from S3We start with a CSV file on S3, which we move through the head node to HDFS and than parse into a spark RDD.  Moving a file from S3 to the head node<jupyter_code><div class="mark">
%cd /mnt/workspace/edX-Micro-Master-in-Data-Science/big-data-analytics-using-spark/notebooks/Data/
!ls</div><i class="fa fa-lightbulb-o "></i>
#create directory to hold data one node
!mkdir Weather
%cd Weather/
#list files on S3
!aws s3 ls s3://dse-weather/ALL.csv.gz
#compressed file is about 1.5GB
%%time
#copy file from S3
!aws s3 cp s3://dse-weather/ALL.csv.gz ./ALL.csv.gz
%%time
#unompress file
!rm ALL.csv
!gunzip ALL.csv.gz
!ls -l ALL.csv
# About 7.7 GB
!head -2 ALL.csv<jupyter_output>station,year,measurement,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,116,117,118,119,120,121,122,123,124,125,126,127,128,129,130,131,132,133,134,135,136,137,138,139,140,141,142,143,144,145,146,147,148,149,150,151,152,153,154,155,156,157,158,159,160,161,162,163,164,165,166,167,168,169,170,171,172,173,174,175,176,177,178,179,180,181,182,183,184,185,186,187,188,189,190,191,192,193,194,195,196,197,198,199,200,201,202,203,204,205,206,207,208,209,210,211,212,213,214,215,216,217,218,219,220,221,222,223,224,225,226,227,228,229,230,231,232,233,234,235,236,237,238,239,240,241,242,243,244,245,246,247,248,249,250,251,252,253,254,255,256,257,258,259,260,261,262,263,264,265,266,267,268,269,270,271[...]<jupyter_text>Distribute file into HDFScopy file from the head-node file system to HDFS<jupyter_code>%%time
!hadoop fs -mkdir /weather
%%time
#create a data directory on hdfs
!hadoop fs -copyFromLocal ALL.csv hdfs:///weather/weather.csv
!hadoop fs -ls /weather<jupyter_output>/bin/sh: hadoop: command not found<jupyter_text>Read csv file into an RDD<jupyter_code>%cd /mnt/workspace/edX-Micro-Master-in-Data-Science/big-data-analytics-using-spark/notebooks/Section2-PCA/PCA/data_preparation/ 
!ls lib
%pwd
!ls -l lib/numpy_pack.py
%%time
%pwd
from pyspark import SparkContext
sc = SparkContext(pyFiles=['/mnt/workspace/edX-Micro-Master-in-Data-Science/big-data-analytics-using-spark/notebooks/Section2-PCA/PCA/data_preparation/lib/numpy_pack.py'])
%%time
RDD=sc.textFile('/weather/weather.csv')
fs_file="/mnt/workspace/edX-Micro-Master-in-Data-Science/big-data-analytics-using-spark/notebooks/Data/Weather/ALL.csv"
!ls -l $fs_file
%%time
with open(fs_file,'r') as f:
    text=f.readlines()
print(len(text))<jupyter_output>9358395
CPU times: user 2.21 s, sys: 5.12 s, total: 7.33 s
Wall time: 7.32 s<jupyter_text>Code for packing and unpacking byte arrays<jupyter_code>import numpy as np
"""Code for packing and unpacking a numpy array into a byte array.
   the array is flattened if it is not 1D.
   This is intended to be used as the interface for storing 
   
   This code is intended to be used to store numpy array as fields in a dataframe and then store the 
   dataframes in a parquet file.
"""
def packArray(a):
    """
    pack a numpy array into a bytearray that can be stored as a single 
    field in a spark DataFrame

    :param a: a numpy ndarray 
    :returns: a bytearray
    :rtype:

    """
    if type(a)!=np.ndarray:
        raise Exception("input to packArray should be numpy.ndarray. It is instead "+str(type(a)))
    return bytearray(a.tobytes())
def unpackArray(x,data_type=np.float16):
    """
    unpack a bytearray into a numpy.ndarray

    :param x: a bytearray
    :param data_type: The dtype of the array. This is important because if determines how many bytes go into each entry in the array.
    :returns: a numpy array
    :rtype: a numpy ndarray of dtype data_type.

    """
    return np.frombuffer(x,dtype=data_type)<jupyter_output><empty_output><jupyter_text>range valuesUsing code that was removed we find that the range of values is `-1000.0, 97892.0` which means that as ints we will need 32 but, but with float we can use just 16.<jupyter_code>#main parsing code

import numpy as np
def parse_weather(line):
    L=line.split(',')
    try:
        assert len(L)==368
        i=2
        L[i]=int(L[i])
        for i in range(3,368):
            if L[i]!='':
                L[i]=np.float16(L[i])
            else:
                L[i]=np.nan
    except:
        #if error in parsing, return (1, input line)
        return (1,line)
    Out=L[:3]
    Out.append(packArray(np.array(L[3:],dtype=np.float16)))
    # if parsing OK, return (0, parsed data)
    return (0,Out)
#this cell demonstrates how to test the parse_weather function on an individual row.
Debug=False
if Debug:
    lines=RDD.take(10)
    GG=parse_weather(lines[-2])
    GG
%%time
Parsed=RDD.map(parse_weather).cache() # filter out bad rows which are mapped (1,line)
DATA=Parsed.filter(lambda x:x[0]==0).map(lambda x:x[1])
ERRORS=Parsed.filter(lambda x:x[0]==1).map(lambda x:x[1])
print(DATA.toDebugString().decode())
%%time
PRCP=DATA.filter(lambda row:row[1]=='PRCP')
print('PRCP records:',PRCP.count())
%%time
print('bad records:',ERRORS.count())
#all lines: 9358395
# only the first line (the header) is bad.
# Good lines: 9358394
DATA.take(1)<jupyter_output><empty_output><jupyter_text>Transform RDD into a Spark DataFrame<jupyter_code>import os
import sys

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import Row, StructField, StructType, StringType, IntegerType, BinaryType, FloatType

# Just like using Spark requires having a SparkContext, using SQL requires an SQLContext
sqlContext = SQLContext(sc)
sqlContext
### Defining the Schema explicitly
# The advantage of creating a DataFrame using a pre-defined schema allows the content of the RDD to be simple tuples, rather than rows.

# In this case we create the dataframe from an RDD of tuples (rather than Rows) and provide the schema explicitly
# Schema with two fields - person_name and person_age
schema = StructType([StructField("Station",     StringType(), True),
                     StructField("Measurement", StringType(), True),
                     StructField("Year",        IntegerType(),True),
                     StructField("Values",      BinaryType(),True)
                    ])
schema
%%time
# Create a DataFrame by applying the schema to the RDD and print the schema
ALL_DataFrame = sqlContext.createDataFrame(DATA, schema)
ALL_DataFrame.printSchema()<jupyter_output>root
 |-- Station: string (nullable = true)
 |-- Measurement: string (nullable = true)
 |-- Year: integer (nullable = true)
 |-- Values: binary (nullable = true)

CPU times: user 460 ms, sys: 0 ns, total: 460 ms
Wall time: 1.29 s<jupyter_text>Write out data frame into Parquet directory<jupyter_code>%%time
!hadoop fs -rm -r /weather/weather.parquet
%%time 
outfilename="hdfs:///weather/weather.parquet"
ALL_DataFrame.write.save(outfilename)
!hadoop fs -du /weather/<jupyter_output>7668890105  /weather/weather.csv
2302800781  /weather/weather.parquet<jupyter_text>Copy parquet directory to head node and then to S3<jupyter_code>%cd /mnt/workspace/Data/
!rm -rf weather.parquet/
!ls -lrt
%%time
!hadoop fs -copyToLocal /weather/weather.parquet weather.parquet
!du .
%%time
#rm parquet directory from s3
!aws s3 rm --recursive --quiet s3://dse-weather/weather.parquet
%%time
# Copy parquet directory from headnode to s3
!aws s3 cp --recursive --quiet ./weather.parquet s3://dse-weather/weather.parquet<jupyter_output>CPU times: user 344 ms, sys: 208 ms, total: 552 ms
Wall time: 18.1 s<jupyter_text>Loading and using a parquet file<jupyter_code>!ls
!rm -rf weather.parquet/
%%time
!aws s3 cp --recursive --quiet s3://dse-weather/weather.parquet ./weather.parquet
%%time
!hadoop fs -copyFromLocal  weather.parquet /weather/weather.parquet
%%time
parquet_name='/weather/weather'
query="""SELECT station,measurement,year 
FROM parquet.`%s.parquet` 
WHERE measurement=\"PRCP\" """%parquet_name
print(query)
df2 = sqlContext.sql(query)
print 'number of rows=',df2.count()
df2.show(5)<jupyter_output>SELECT station,measurement,year 
FROM parquet.`/weather/weather.parquet` 
WHERE measurement="PRCP" 
number of rows= 2521007
+-----------+-----------+----+
|    station|measurement|year|
+-----------+-----------+----+
|IN011020501|       PRCP|1939|
|IN011020501|       PRCP|1940|
|IN011020501|       PRCP|1941|
|IN011020501|       PRCP|1942|
|IN011020501|       PRCP|1943|
+-----------+-----------+----+
only showing top 5 rows

CPU times: user 12 ms, sys: 0 ns, total: 12 ms
Wall time: 2.59 s
{'fraction_non_alphanumeric': 0.13250954390194897, 'fraction_numerical': 0.094333936106088, 'mean_word_length': 4.460778935820077, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '101075', 'license': 'Apache-2.0', 'n_tokens_mistral': 4099, 'n_tokens_neox': 3418, 'n_words': 998, 'path': 'Sections/Section1-Spark-Basics/2.SparkSQL/3.S3_HDFS_Parquet.ipynb', 'repo_name': 'sookieliao/UCSD_18_Spring_Big-Data-Analytics-using-Spark-Implementation-', 'size': '9954'}