Home
The main aim of this project is to use Chained Hadoop MapReduce jobs to analyse historical weather and wildfire activities data in Australia. We explore the impact of temperature on the strength of wildfires and the extent of burnt areas. By leveraging the power of Hadoop MapReduce, we can efficiently process large-scale datasets and gain insights into this critical issue.
Wildfires are uncontrollable fires that occur in green vegetation and wild areas, such as forests and farms. They have been affecting green areas worldwide for many years, causing significant harm to humans, trees, and animals. This project focuses on analysing historical weather and wildfire activities data collected in Australia between 2005 and 2021.
The project utilises two datasets from the IBM wildfire data repository. [Predicting wildfires with weather forecast data url: (https://developer.ibm.com/exchanges/data/all/spot-challenge-wildfires/) .]
Historical Wildfires: This dataset contains estimated measures for wildfire activities. The data was collected and processed by NASA's Earth Science Data and Information System (ESDIS) using satellite data. Each record represents a wildfire event, labelled by date and aggregated to a daily level. The dataset includes variables such as fire area and fire brightness.
Historical Weather: This dataset consists of measures collected from hourly ERAS climate reanalysis and processed using IBM PAIRS Geoscope. The data is aggregated to a daily level for each region and includes weather conditions such as temperature, wind speed, radiation, and humidity.
Both datasets cover seven regions in Australia: New South Wales (NSW), Northern Territory (NT), Queensland (QL), South Australia (SA), Tasmania (TA), Victoria (VI), and Western Australia (WA).
By studying the factors that contribute to the initiation, spread, and intensity of wildfires, we aim to answer the following research question:
- Do higher temperatures make wildfires stronger and cause larger burnt areas in any particular region?
To answer this question, we need to:
- Find the correlation between the maximum temperature and the burnt area, and the correlation between the maximum temperature and the fire brightness (this is one of the measurements to determine the fire strength).
- See how the fire strength and burnt area change with time to determine if they increase during the times when the temperatures are the highest.
The design of the project consists of four chained MapReduce jobs as is shown in figure Figure 1. The first two jobs are multi-step MapReduce jobs that work in parallel to process the two data files separately. Once the first two jobs are completed, a third dependent MultiInputs MapReduce job is triggered. The main purpose of this job is to join the outputs of the first two jobs and emit a single data file. When the third job finishes, a fourth dependent job begins, which focuses on finding Pearson's Correlation Coefficients between the numerical variables.
In this project, we have implemented a series of four chained MapReduce jobs to analyse the historical weather and wildfire data. The execution order and dependencies of these jobs are controlled by `JobControl'.
Each individual multi-step job consists of multiple mappers and a reducer. The execution order of the mappers and reducer within a job follows the order specified in the code. The chaining of mappers and reducers is managed using ChainMapper and ChainReducer, ensuring that the data flows seamlessly through each step of the job.
To combine the data files, we utilise the MultipleInputs functionality, which allows us to pass multiple inputs to the MapReduce job. This enables us to merge the outputs of the previous jobs and process them together in subsequent steps.
The first two jobs are designed to process the Historical Weather and Historical Wildfire data files separately. Each job consists of multiple steps executed in a sequential manner. These steps involve filtering the data, extracting relevant variables, and performing calculations. The output of these jobs serves as input for the subsequent stages.
In the first job of processing the historical weather data, the focus is on the Temperature parameter. To achieve this, the parameter values are filtered to include only temperature. Additionally, the effect of temperature on wildfires is studied for each region individually, necessitating another filter based on the region. The temperature filter will remains fixed in the mapper class, while the region filter is determined by the user input in args[5], representing one of the seven Australian regions in the data.
The first mapper in Job 1 (WeatherLocationFilterMappaer.java) includes the filtering stage and extracts the other variables of interest, namely the date and the maximum temperature max(). The mapper emits the region as a Text key and (date, max.temperature) as a Text value.
The output of the first mapper is then taken as input for the second mapper (WeatherDateMapper.java). It is important to note that this is a second mapper, not a reducer, which means the records in the file are read row by row, similar to a text file, with variables separated by tabs and commas. This mapper modifies the date by extracting the year and month, and sets (date, location) as a Text key, and converts the max.temperature to a DoubleWritable value. During the shuffle and sort phase, the temperatures (values) for each month in every year (keys) are grouped together and allocated to the same reducer. This arrangement facilitates the computation of the mean temperature for each year-month combination.
The input for the reducer(WeatherReducer.java), the input consists of a Text key representing a certain month in a specific year, and a DoubleWritable array containing all the recorded temperatures for that month. The reducer iterates through the array, sums all the values, and calculates the average by dividing the sum by the count of temperature records in that month (array size). Consequently, the average temperature for each month is obtained. The reducer emits the same input Text key (date[YYYY-mm], location) and a DoubleWritable value representing the mean maximum temperature.
The chaining of mappers and reducers within the first job is controlled by the driver class in (driver.java). Default settings for the mapper and reducer functions are overridden, and ChainMapper.addMapper() and ChainReducer.setReducer() methods are employed to determine the execution order, input, and output for each step in the job.
For a more comprehensive understanding, please see the commented Java codes for the first multi-step job. The structure of inputs and outputs within the job is described in the table below.
| Stage | Input | Input Type | Output | Output Type |
|---|---|---|---|---|
| Mapper1 Filters(region and Temp) |
Data set and Row number | K: LongWritable V: Text |
Region, (Date,MaxTemp) Eg: NSW, (2005-01-13, 5.03 ) |
K: Text V: Text |
| Mapper2 Reformat the date |
Region, (Date,MaxTemp) Region,(YYYY-mm-dd, Temp) |
K: Text V:Text |
(Date,Region),MaxTemp (YYYY-mm, %S), 0.0 |
K:Text V:DoubleWritable |
| Reducer Finds the mean of MaxTemp for each year-month |
(YYYY-mm,Region) and list of MaxTemp values for each day of the month in Key | K: Text V: DoubleWritable |
(YYYY-mm), mean(MaxTemp) for the values in the input list | K:Text V:DoubleWritable |
The second MapReduce job is dedicated to processing the Historical wildfire data. Our objective is to extract the Estimated fire area and Mean estimated fire brightness variables from the data file, along with the Date and Region information.
The processing steps for the HistoricalWildfire dataset are identical to those of the first job. However, in this case, we need to calculate the mean values for two variables instead of one: the fire area and fire brightness. Additionally, the region filtering in the first mapper will take the same region specified in the command line input, which corresponds to the same region used in the first MapReduce job (fifth argument).
The commented Java codes for the second multi-step job can be found in (HistoricalFiresMapper.java, HistoricalFiresDateMapper.java and HistoricalWildFiresReducer.java)
To manage the chaining of mappers and reducer, we rely on the driver class, which controls the execution flow. This is similar to the approach discussed earlier. Detailed information regarding the inputs and outputs formats for each stage of this second multi-step MapReduce job can be found in the table below.
| Stage | Input | Input Type | Output | Output Type |
|---|---|---|---|---|
| Mapper1 Filters the region |
Data set and Row number | K: LongWritable V: Text |
Region, (Date,FireArea,FireBrightness) Eg: NSW, ((01/13/2005, 5.03, 300.1) |
K: Text V: Text |
| Mapper2 Reformat the date |
Region, (Date, Area, FireBrightness) Region,(mm-dd-YYYY, 0.0, 0.0) |
K: Text V:Text |
(Date,Region),(Area, FireBrightness) [(YYYY-mm, %S), (0.0, 0.0) |
K:Text V: Text |
| Reducer Finds the mean of FireArea and FireBrightness for each year-month |
(YYYY-mm, Region) and list of Text tuples (Area, Brightness) values for each day of the month in Key | K: Text V: Text |
(YYYY-mm), mean(Area, Brightness) for the values in the input list | K:Text V: Text |
Once we have processed each data file individually, the next step is to combine them to examine the relationship between their variables. This is achieved through the third MapReduce job, referred to as the "Joining" job. In this job, the MultipleInputs.addInputPath() method in the driver class enables access to multiple inputs for the MapReduce job.
It is important to note that while the previous two jobs run concurrently (simultaneously), the Joining job operates in a sequential manner. This sequential execution is enforced by setting dependencies in the JobControl within the driver class, ensuring that the Joining job runs after job1 and job2 have completed.
The Joining job starts by running two mappers in parallel. The purpose of these mappers is to label the values from the outputs of the previous MapReduce jobs. The first mapper labels the values from the "weather" job output by adding "T:" at the beginning. On the other hand, the second mapper labels the values from the "HistoricalFires" job output by adding "F:" at the start of each value. This labelling process is crucial for ensuring that the variables are joined in the same order consistently. For example, (Temperature, Area, Fire Brightness) will always have the same order for each merged value.
It is worth noting that the input for each mapper is treated as a text file, read row by row, with the inputs separated by some character. In this case, the input is obtained from another MapReduce output, where the keys and values are separated by tabs. However, after splitting on the tab, we can handle the inputs as usual, with values separated by commas.
During the sorting and shuffling phase, the output values from each mapper are grouped based on the common key. For example:
If mapper1 (histo_Wildfire_join.java) emits → [(2005-01, NSW), T:30.2]
and mapper2 (Join_weather_mappr.java) emits → [(2005-01, NSW), F:8.5, 301.7]
Then in sorting and shuffling it becomes → [(2005-01,NSW), {(T:30.2), (F:8.5, 301.7)}]
The reducer waits for the results from the two mappers before commencing its functionality. The main task of the reducer is to iterate through the values for each key. If the value starts with "T:", the reducer places it on the left side of the values and if the value starts with "F:", it allocates it to the right. This approach ensures that the values for each variable always have the same index on the same line. Referring back to the previous example:
the reducer (JoiningReducer.java) emits [(2005-01, NSW), (30.2, 8.5, 301.7)].
Thus, the emitted value consistently follows the format (Temp, Area, Brightness), which is essential for consistency and further processing.
Throughout the entire process, the Region information is retained in the data to facilitate better referencing based on the user's choice. However, it could have been removed in the first job if desired. The following table represents the structure of inputs and outputs for the mappers and reducers in the third job. The output of this reducer serves as the initial output of interest for answering the research question. The output directory can be determined by the fourth argument entered during the execution.
| Stage | Input | Input Type | Output | Output Type |
|---|---|---|---|---|
| Mapper1 Labels values with (T:) |
Data file and Row number | K: LongWritable V: Text |
[(Date,Region), T:AvgTemp] | K: Text V: Text |
| Mapper2 Labels values with (F:) |
Data file and Row number | K: LongWritable V:Text |
[(Date,Region), (F:AvgTemp, AvgBrightness]] | K:Text V: Text |
| Reducer Joins the mappers output by common key |
key is (date,region) and the values are list of Text tuples labelled by the data source ("F:area,brightness", "T:Temp") | K: Text V: Text |
(Date,Region), (Temp,Area,Brightness) | K:Text V: Text |
The purpose of the fourth MapReduce job is to calculate the Pearson correlation coefficient, which measures the linear dependency between two features. The coefficient ranges from 1 to -1, indicating the strength and direction of the linear relationship. A coefficient closer to +1 signifies a strong positive linear relationship, while a coefficient closer to -1 indicates a strong negative linear relationship. A coefficient near 0 suggests no linear dependency between the variables.
The reason for performing this MapReduce job is to determine if the wildfire strength (fire brightness and fire area) is linearly dependent on the weather temperature. The Pearson correlation coefficient between two variables (x and y) can be calculated using the following equation:
where r is Pearson's correlation coefficient.
and the equation can be written as,
where,
To calculate the correlation coefficient in hadoop mapreduce, the second equation is used. he approach involves breaking down the equation into smaller parts such as,
Based on the previous MapReduce job output, which consisted of (Temperature, Area, Fire Brightness), we aim to calculate three correlation coefficients:
The correlation coefficients we want to calculate are between:
(Temperature and Area), (Temperature and Fire brightness) and (Area and Fire Brightness). Now, if we expressed the coefficients that we need in terms of the values' indices in each row that would be like:
(Temperature,Area)
(Temperature,Fire Brightness)
(Area,Fire Brightness)
where,
0: Temperature index in each row,
1: Area index in each row,
2: Fire Brightness index in each row
Next, let's discuss the mapper's role in this process (PearsoCorrMApper.java). The primary task of the mapper is to read each line of data and return a key that contains the indices for each pair of variables along with a value that had the real measurements at these indices. During the sorting and shuffling phase, all pairs with the same indices will be grouped together. Let's consider a simple example to illustrate this process:
Assume we have the following table:
| row index | Temp | Area | Brightness |
|---|---|---|---|
| 1 | 30.2 | 20.5 | 150.8 |
| 2 | 25.1 | 40 | 120 |
| 3 | 27 | 10 | 400 |
Then, the mapper's output for the first row will be:
K: (0,1)
K: (0,2)
K: (1, 2)
And, the mapper output for the second row will be:
K: (0,1)
K: (0,2)
K: (1, 2)
and for the third row:
K: (0,1)
K: (0,2)
K: (1, 2)
During the sorting and shuffling phase, pairs with the same indices will be grouped together, resulting in the following intermediate output:
K: (0,1)
K: (0,2)
K: (1,2)
This grouping allows the reducer (PearsonCorrReducer.java) to process each group of values for the same pair of indices, enabling the calculation of correlation coefficients. The table below shows the structure of the inputs and outputs for the job.
| Stage | Input | Input Type | Output | Output Type |
|---|---|---|---|---|
| Mapper1 index the the pairs for each correlation coefficient |
Data file and Row number | K: LongWritable V: Text |
the keys are the indices pairs for each two variables and the values are the measurements in those indices | K: Text V: Text |
| Reducer finds Pearson's corr coefficients (r) |
[(index1,index2), (value1, value2)] K: (i,j) V: (x,y) |
K: Text V: Text |
The index as a key and the correlation between every two variables. | K:Text V: DoubleWritable |
In the driver class, which is responsible for controlling the execution flow of the MapReduce jobs, the chaining of jobs and their execution are handled. This section focuses on the code snippets for chaining the jobs and running them.
Firstly, the driver class checks the number of arguments provided in the command prompt. In this project, it expects six arguments: the driver class itself (wildFire), the input path for the historical weather text file, the input path for the historical wildfires text file, the output directory for the Joining job output, the output directory for the correlation output, and the selected region (case insensitive) in Australia (NSW, NT, QL, SA, TA, VI, or WA). The command prompt instructions are shown in the next figure.
After that, the settings for each MapReduce job are declared as discussed in the earlier sections. Additionally, the settings for controlling the job chaining process are introduced. Each input for the MapReduce jobs is treated as a new text file, even if it is an output of a previous job. Therefore, all job inputs are of type LongWritable.
The dependencies between jobs are set using ControlledJob(). For example, job3 depends on job1 and job2, so it will not start until job1 and job2 finish their execution. The dependencies are passed as an ArrayList to the method, even if there is only one dependency. If there are no dependencies, the ArrayList is replaced with null. The chaining process itself is controlled by JobControl, where a JobControl instance is initialized for the chaining process. In this case, there is one process that controls all jobs, named "wildFire MRJobs". All the ControlledJobs are added to the JobControl.
To run the chained jobs, Thread() is used. It starts the defined JobControl and continuously checks if all the jobs have been executed. Once all the jobs have finished, the code exits. The code snippet bellow shows the libraries used in the driver class.
and the next code shows the Java code snippets for JobControl() and Thread() (for more detailed code you can refer to the code files).
In this tutorial, we have successfully implemented a series of four chained MapReduce jobs to analyse wildfire data and its relationship with weather conditions. By breaking down the problem into smaller tasks and chaining them together, we have achieved several advantages in terms of manageability and performance.
Dividing the problem into smaller sub-problems allowed us to focus on specific tasks in each job, making development and debugging easier. Moreover, by keeping the output of mappers in memory instead of storing it on disk, we reduced disk I/O operations, leading to improved efficiency.
However, it is essential to be mindful of memory usage, especially when working with large datasets. As the size of mapper outputs can become substantial, proper memory management becomes crucial to ensure smooth execution.
Looking ahead, for those interested in more advanced tasks, an exciting avenue to explore is utilising the entire wildfire dataset and modifying the code to analyse all available weather conditions. This expanded analysis would provide a more comprehensive understanding of the relationship between weather conditions and wildfires, offering valuable insights for further research.