# TP1 BDA
**Kevin Butikofer, Charles-Lewis Jaggi**
**Groupe : L1 GrK**
## TASK 1
### Running the sample programs

## TASK 2
### EMR Console

### Cost
- 

- **Trois m4.large instance.**

Le temps écoulé est de 9 minutes et on a utlisé 3 instances donc :
-> $(0.06 + 3 * 0.2 )*(1/60 *9) = 0.099$ $

### key-value pairs process

Job JOBID="job_202002281331_0001" FINISH_TIME="1582897008124" 
JOB_STATUS="SUCCESS" FINISHED_MAPS="215" FINISHED_REDUCES="7" 
FAILED_MAPS="3" FAILED_REDUCES="1" 
MAP_COUNTERS="{(org\.apache\.hadoop\.mapred\.FileInputFormat$Counter)
(File Input Format Counters )[(BYTES_READ)(Bytes Read)(85543660)]}
{(FileSystemCounters)(FileSystemCounters)[(FILE_BYTES_WRITTEN)
(FILE_BYTES_WRITTEN)(12696448)][(HDFS_BYTES_READ)
(HDFS_BYTES_READ)(24080)][(S3_BYTES_READ)(S3_BYTES_READ)(85543660)]}
{(org\.apache\.hadoop\.mapred\.Task$Counter)(Map-Reduce Framework)
[(COMBINE_INPUT_RECORDS)(Combine input records)(0)]
[(COMBINE_OUTPUT_RECORDS)(Combine output records)(0)]
[(COMMITTED_HEAP_BYTES)(Total committed heap usage \\(bytes\\))(211067863040)]
[(CPU_MILLISECONDS)(CPU time spent \\(ms\\))(71800)]
[(MAP_INPUT_BYTES)(Map input bytes)(651778167)]
[(MAP_INPUT_RECORDS)(Map input records)(2831380)]
[(MAP_OUTPUT_BYTES)(Map output bytes)(31031858)]
[(MAP_OUTPUT_MATERIALIZED_BYTES)(Map output materialized bytes)(7017005)]
[(MAP_OUTPUT_RECORDS)(Map output records)(2821078)]
[(PHYSICAL_MEMORY_BYTES)(Physical memory \\(bytes\\) snapshot)(202894118912)]
[(SPILLED_RECORDS)(Spilled Records)(2821078)]
[(SPLIT_RAW_BYTES)(SPLIT_RAW_BYTES)(24080)]
[(VIRTUAL_MEMORY_BYTES)(Virtual memory \\(bytes\\) snapshot)(442712690688)]}
" REDUCE_COUNTERS="{(org\.apache\.hadoop\.mapred\.FileOutputFormat$Counter)(File Output Format Counters )
[(BYTES_WRITTEN)(Bytes Written)(387)]}
{(FileSystemCounters)(FileSystemCounters)
[(FILE_BYTES_READ)(FILE_BYTES_READ)(6991478)]
[(FILE_BYTES_WRITTEN)(FILE_BYTES_WRITTEN)(7175154)]
[(S3_BYTES_WRITTEN)(S3_BYTES_WRITTEN)(387)]}
{(org\.apache\.hadoop\.mapred\.Task$Counter)(Map-Reduce Framework)
[(COMBINE_INPUT_RECORDS)(Combine input records)(0)]
[(COMBINE_OUTPUT_RECORDS)(Combine output records)(0)]
[(COMMITTED_HEAP_BYTES)(Total committed heap usage \\(bytes\\))(1734868992)]
[(CPU_MILLISECONDS)(CPU time spent \\(ms\\))(12240)]
[(PHYSICAL_MEMORY_BYTES)(Physical memory \\(bytes\\) snapshot)(1247551488)]
[(REDUCE_INPUT_GROUPS)(Reduce input groups)(43)]
[(REDUCE_INPUT_RECORDS)(Reduce input records)(2821078)]
[(REDUCE_OUTPUT_RECORDS)(Reduce output records)(43)]
[(REDUCE_SHUFFLE_BYTES)(Reduce shuffle bytes)(7017005)]
[(SPILLED_RECORDS)(Spilled Records)(2821078)]
[(VIRTUAL_MEMORY_BYTES)(Virtual memory \\(bytes\\) snapshot)(14553698304)]}"
 COUNTERS="{(org\.apache\.hadoop\.mapred\.FileInputFormat$Counter)(File Input Format Counters )
 [(BYTES_READ)(Bytes Read)(85543660)]}
 {(org\.apache\.hadoop\.mapred\.JobInProgress$Counter)
 (Job Counters )[(FALLOW_SLOTS_MILLIS_MAPS)(Total time spent by all maps waiting after reserving slots \\(ms\\))(0)]
 [(FALLOW_SLOTS_MILLIS_REDUCES)
 (Total time spent by all reduces waiting after reserving slots \\(ms\\))
 (0)][(RACK_LOCAL_MAPS)(Rack-local map tasks)(218)][(SLOTS_MILLIS_MAPS)(SLOTS_MILLIS_MAPS)(947587)]
 [(SLOTS_MILLIS_REDUCES)(SLOTS_MILLIS_REDUCES)(460841)][(TOTAL_LAUNCHED_MAPS)(Launched map tasks)(218)]
 [(TOTAL_LAUNCHED_REDUCES)(Launched reduce tasks)(8)]}
 {(org\.apache\.hadoop\.mapred\.FileOutputFormat$Counter)
 (File Output Format Counters )
 [(BYTES_WRITTEN)(Bytes Written)(387)]}
 {(FileSystemCounters)(FileSystemCounters)[(FILE_BYTES_READ)(FILE_BYTES_READ)(6991478)]
 [(FILE_BYTES_WRITTEN)(FILE_BYTES_WRITTEN)(19871602)][(HDFS_BYTES_READ)(HDFS_BYTES_READ)(24080)]
 [(S3_BYTES_READ)(S3_BYTES_READ)(85543660)][(S3_BYTES_WRITTEN)(S3_BYTES_WRITTEN)(387)]}
 {(org\.apache\.hadoop\.mapred\.Task$Counter)
 (Map-Reduce Framework)
 [(COMBINE_INPUT_RECORDS)(Combine input records)(0)]
 [(COMBINE_OUTPUT_RECORDS)(Combine output records)(0)]
 [(COMMITTED_HEAP_BYTES)(Total committed heap usage \\(bytes\\))(212802732032)]
 [(CPU_MILLISECONDS)(CPU time spent \\(ms\\))(84040)][(MAP_INPUT_BYTES)(Map input bytes)(651778167)]
 [(MAP_INPUT_RECORDS)(Map input records)(2831380)]
 [(MAP_OUTPUT_BYTES)(Map output bytes)(31031858)]
 [(MAP_OUTPUT_MATERIALIZED_BYTES)(Map output materialized bytes)(7017005)]
 [(MAP_OUTPUT_RECORDS)(Map output records)(2821078)]
 [(PHYSICAL_MEMORY_BYTES)(Physical memory \\(bytes\\) snapshot)(204141670400)]
 [(REDUCE_INPUT_GROUPS)(Reduce input groups)(43)][(REDUCE_INPUT_RECORDS)(Reduce input records)(2821078)]
 [(REDUCE_OUTPUT_RECORDS)(Reduce output records)(43)]
 [(REDUCE_SHUFFLE_BYTES)(Reduce shuffle bytes)(7017005)]
 [(SPILLED_RECORDS)(Spilled Records)(5642156)]
 [(SPLIT_RAW_BYTES)(SPLIT_RAW_BYTES)(24080)]
 [(VIRTUAL_MEMORY_BYTES)(Virtual memory \\(bytes\\) snapshot)(457266388992)]}

[(MAP_INPUT_RECORDS)(Map input records)(2831380)]

Mappers Input : 2'831'380

[(REDUCE_INPUT_RECORDS)(Reduce input records)(2821078)]

Reducers Input : 2'821'078

### Temparture Chart

Overall highest temperature : 38.0° en 2003

## Task 3

- La température de 22° a 56'530 occurences
- La température minimal est de -25° et la température maximal est de 38°
- La température qui a le plus d'occurence est 13° avec 114'613


### Mapper

#!/usr/bin/env python
#
# histo_temperature_map.py - Calculate histogram temperature from NCDC Global
#                          Hourly Data - Mapper part

import re   # import regular expressions
import sys  # import system-specific parameters and functions

# loop through the input, line by line
for line in sys.stdin:
  # remove leading and trailing whitespace
  val = line.strip()
  # extract values for year, temperature and quality indicator
  year = val[15:19]
  temp = val[87:92]
  q = val[92:93]
  # temperature is valid if not +9999 and quality indicator is
  # one of 0, 1, 4, 5 or 9
  if (temp != "+9999" and re.match("[01459]", q)):
    print "%s" % (int(float(temp)/10))

### Reducer

#!/usr/bin/env python
#
# histo_temperature_reduce.py - Calculate histogram temperature from NCDC Global
#                             Hourly Data - Reducer part
#import collections
import sys

last_val = None
cpt = 0
# loop through the input, line by line
for line in sys.stdin:
  # each line contains a key and a value separated by a tab character
  val = line.strip()
  # Hadoop has sorted the input by key, so we get the values
  # for the same key immediately one after the other.
  # Test if we just got a new key, in this case output the maximum
  # temperature for the previous key and reinitialize the variables.
  # If not, keep calculating the maximum temperature.
  if last_val and last_val != val:
    print "%s\t%s" % (last_val, cpt)
    cpt = 1
  else:
    cpt += 1
  last_val = val

# we've reached the end of the file, output what is left
if last_val:
  print "%s\t%s" % (last_val, cpt)