Hadoop Phase One Summary
Phase One Summary of Hadoop
Updated:
阅读中文版Hadoop Phase Learning Summary
Part 1: HDFS Related Questions
1. Describe the HDFS data write process
First, the client sends a write data request to the NameNode service. After receiving the request, the NameNode performs basic validation, which includes verifying the legality of the requested upload path and also performing permission verification for the requesting user. If the validation passes, the NameNode responds to the client allowing the upload. Next, the client splits the file into blocks according to the blocksize, and after splitting, uploads them sequentially in units of blocks. At this point, the client requests to upload the first block information. After receiving the upload request, the server returns three DataNode machines for storing data block replicas by default based on HDFS's default rack awareness principle. After receiving the machine list, the client finds one machine to establish a transmission channel based on the network topology principle, and then serially connects to the three machines in sequence. This connection method is mainly to reduce the IO pressure on the client side. When the channel is established successfully, the client transmits data through HDFS's FSOutputStream stream object, with the minimum unit of data transmission being a packet. During transmission, each DataNode server is serially connected, passing the data along in sequence. When the last data block is fully transmitted, one write operation is considered complete. If there are more data blocks to transmit, the second data block is transmitted next.
2. Describe the HDFS data read process
First, similar to writing data, the client sends a request to the NameNode. After receiving the request, the NameNode performs validation of the file download path's legality and permission verification. If the validation passes, it returns the metadata information of the target file to the client, which includes the DataNode location information corresponding to the target file's data blocks. Then, based on the specific DataNode location information combined with the proximity principle and network topology principle, the client finds the server closest to itself to access and download the data. Finally, the data is read locally through the FSInputStream object provided by HDFS. If there are multiple block information entries, the DataNode will be requested multiple times until all data of the target file is downloaded.
3. Briefly describe the HDFS architecture and the role of each service
HDFS is the file system in the Hadoop architecture responsible for distributed data storage management. When a non-high-availability HDFS cluster is running, three services are started: NameNode, DataNode, and SecondaryNameNode. Among these, NameNode is the central service of HDFS, mainly responsible for maintaining and managing the metadata information of files in the file system. DataNode is mainly responsible for storing the actual data block information of files. Of course, the data block information on DataNode also includes some metadata information about the current data block, such as checksum values, data length, timestamps, etc. In a non-high-availability HDFS cluster, NameNode and DataNode can be understood as a one-to-many relationship. The two must also maintain communication during cluster operation, typically checking heartbeats every 3 seconds by default. Finally, the SecondaryNameNode's job is very singular: it is to merge NameNode's metadata image file and edit log, and also keep a copy of the metadata information itself as a recovery guarantee in case NameNode's metadata is lost.
4. How is metadata maintenance implemented in HDFS?
NameNode's metadata information is maintained through the fsimage file + edits edit log. When NameNode starts, the contents of the fsimage file and the edits edit log are loaded into memory and merged to form the latest metadata information. When we operate on metadata, considering the inefficiency of directly modifying files, we do not directly modify the fsimage file, but instead append operation records to the edits edit log file. When certain conditions are met, we let SecondaryNameNode complete the merging of the fsimage file and the edits edit log file. SecondaryNameNode first asks NameNode to stop using the currently active edits edit log file and generates a new edits edit log file. Then it copies NameNode's fsimage file and the stopped edits file locally, merges the operation records from the edits edit log file into the fsimage file in memory to form a new latest fsimage file, and finally pushes this latest fsimage file to NameNode while also keeping a backup copy for itself.
5. Describe the relationship between NN and DN, and the workflow of DN
From a data structure perspective, NameNode and DataNode have a one-to-many relationship. An HDFS cluster can only have one NameNode for maintaining metadata information, while there can be multiple DataNodes for storing actual data blocks. When the HDFS cluster starts, it first enters safe mode. In safe mode, we can only read data and cannot perform any write operations. At this time, each DataNode server in the cluster registers itself with the NameNode. After successful registration, the DataNode reports its detailed data block information. When the data block report satisfies the minimum replica condition, safe mode automatically exits. After that, DataNode and NameNode communicate every three seconds. If NameNode detects that a DataNode has not responded, it continues to check until 10 minutes and 30 seconds have passed without detection, at which point it determines that the current DataNode is unavailable.
Part 2: MapReduce Related Questions
1. Describe the general process and conventions of writing MR by hand
First, from the structural division of a MapReduce program, it can be divided into three parts. The first is the program execution entry, usually referred to as the driver class, which mainly writes the MR job submission process and custom configuration items. The second is the core class of the Map phase, which needs to be customized and inherit the Mapper class provided by Hadoop, override the map method in the Mapper class, write your own business logic code in the map method, and write the processed data out to disk using the context object. The third is the core class of the Reduce phase, which also needs to inherit the Reducer class provided by Hadoop and override the reduce method, writing your own business logic code in the reduce method, and writing the data out through the context object after processing, which is the final result file.
2. How to implement serialization in Hadoop, and what is the difference between Hadoop serialization and Java serialization?
First, serialization is the process of converting Java objects in memory into binary bytecode, and deserialization is the process of converting binary bytecode into Java objects. Usually, we need to perform serialization when persisting Java objects to disk or transmitting Java objects as data over the network. Conversely, if we want to read data from disk and convert it into Java objects, we need to perform deserialization. To implement serialization in Hadoop, the JavaBean object needs to implement the Writable interface and override the write() method and readFields() method, where the write() method is the serialization method and the readFields() method is the deserialization method.
The difference between Hadoop serialization and Java serialization is that Java serialization is more heavyweight. The result of Java serialization not only generates binary bytecode files but also generates corresponding verification information and integration architecture for the current Java object. This means we need to maintain more data unnecessarily. However, Hadoop serialization does not produce any information other than the internal properties of the Java object. The overall content is more concise and compact, and the read/write speed is correspondingly much faster, which also fits the context of big data processing.
3. Outline the execution flow of an MR program
To describe it simply, the MR program execution starts with the InputFormat class. InputFormat is responsible for data reading and performs splitting internally. Each split of data corresponds to generating one MapTask. In the MapTask, data is processed line by line according to the file's lines. Each line of data calls the map method of our custom Mapper class once. The map method implements the specific business logic internally. After processing the data, it writes the data out to disk through the context object (this goes through the Shuffle process - see question 7 below for details!). Next, the ReduceTask begins execution. First, the ReduceTask copies the data results processed by the MapTask. Each group of values with the same key calls the reduce method of our custom Reducer class once. When data processing is complete, the data results are written out to disk through the context object.
4. InputFormat performs splitting when writing data. Why is the default split size 128M?
First, the split size can be changed by modifying configuration parameters, but by default it is consistent with the blocksize. The purpose of this is to be able to read exactly one block of data at a time when reading data, avoiding cross-machine reading in a cluster environment. If cross-machine reading occurs, it would cause additional network IO, which is not conducive to improving the execution efficiency of MR programs.
5. Describe the splitting logic (from a source code perspective)
Splitting in MR occurs during the data reading phase, so we need to focus on the implementation of InputFormat. By tracing the source code, there is a getSplits() method in the InputFormat abstract class, which is the specific logic for implementing splitting. First, we focus on two variables: minSize and maxSize. By tracing the source code, by default minSize = 1 and maxSize = Long.MAX_VALUE. The source code declares a collection List
To increase the split size, modify mapreduce.input.fileinputformat.split.minsize. To decrease the split size, modify mapreduce.input.fileinputformat.split.maxsize.
Once we can obtain the split size, we can continue executing. Before finally completing the splitting, there is a key judgment: determining whether the remaining file should continue to be split. If the remaining file/split size > 1.1, then continue splitting; otherwise, no more splitting will be performed. This rule considers making future splits as resource-balanced as possible, so that very small file contents don't also start a MapTask. This completes the entire splitting rule description!
6. How is the CombineTextInputFormat mechanism implemented?
CombineTextInputFormat is also an implementation class of InputFormat, mainly used to solve small file scenarios. If we are processing a large number of small files, since the default splitting rule splits based on individual files, this leads to the generation of a large number of MapTasks, but each MapTask processes a very small file, which goes against the original design intent of MapReduce. If we encounter the above scenario, we cannot use the default splitting rule but instead use the splitting rule in CombineTextInputFormat. The general idea of the splitting rule in CombineTextInputFormat is: first, set a parameter for the maximum split value when submitting the Job. Once this value is set and the InputFormat implementation class is specified as CombineTextInputFormat in the Job submission, then during the splitting process, the current file size is first compared with the set maximum split value. If it is smaller than the maximum split value, it is independently divided into one block. If it is larger than the maximum split value but smaller than twice the maximum split value, the current file is divided into two blocks. This process continues for each file, and this process is called the virtual process. Finally, when generating the actual splits, merging is done based on the virtually divided files. As long as the merged file size does not exceed the initially set maximum split value, files continue to be appended and merged until reaching the set maximum split value. At this point, one split is generated corresponding to one MapTask.
7. Explain the Shuffle mechanism process
Shuffle is a very important and indispensable process in MR execution. When the MapTask finishes executing the map() method and writes data through the context object, the shuffle process begins. First, data is written from the Map side into the ring buffer. The written data enters the designated partition according to the partitioning rules, and at the same time, in-memory sorting is performed within the partition. The default size of the ring buffer is 100M. When the data write capacity reaches 80% of the buffer size, data begins to spill to disk. If there is a large amount of data, multiple spills may occur, resulting in multiple spill files on disk, and each spill file ensures that the data within the partition is ordered. Next, on disk, the multiple spilled files are merged into one file. During this merging process, merge sorting is performed according to the same partition, ensuring that the merged file is ordered within partitions. At this point, the shuffle process on the Map side is complete. Then, the data output from the Map side serves as input data for the Reduce side for further aggregation. At this time, the ReduceTask copies the data from the same partition calculated by each MapTask into the ReduceTask's memory. If memory cannot hold it, it begins writing to disk. Then, merge sorting is performed on the data. After sorting, grouping is performed based on the same key. In the future, one group of values corresponding to the same key will call the reduce method once. If there are multiple partitions, multiple ReduceTasks will be generated to handle them, and the processing logic is the same.
8. In an MR program, who determines the number of partitions, and at which stage does writing data into partitions begin?
In an MR program, from the perspective of code configuration analysis, the number of ReduceTasks can be set when submitting the Job. The number of ReduceTasks determines the partition numbering. By default, as many ReduceTask tasks as there are, that many partitions will be generated. However, how many ReduceTasks should actually be set is determined by the specific business. In the Map phase, when data is written out through context.write() in the map method, it is actually writing data into the designated partition.
9. Explain the approach for implementing partitioning in MR (from a source code perspective)
Partitioning is an important concept in MR. Usually, partitioning is determined by specific business logic. By default, if no partition count is specified, there will be one partition. If we want to specify partitions, we can specify the number of partitions by setting the number of ReduceTasks when submitting the Job. After data is processed on the Map side, it is spilled to the designated partition. Which partition a kv data goes to is determined by the partitioner object provided by Hadoop, called Partitioner. The default implementation of the Partitioner object is the HashPartitioner class. By tracing the source code, when we call the map method to write data out, HashPartitioner is called. Its rule is to perform a modulo operation using the key of the current data being written and the number of ReduceTasks set in the Job submission. The result obtained is the partition number of the partition where the current data should be written. In addition, we can also customize the partitioner object by inheriting the Partitioner object provided by Hadoop, then overriding the getPartition() method, and implementing the return of the partition number according to our own business logic in that method. Finally, we set our custom partitioner object in the Job submission code to override the default partitioning rule.
10. What are the two approaches for implementing sorting in Hadoop?
The first implementation approach is to directly make the objects participating in comparison implement the WritableComparable interface and specify the generic type. Next, implement the compareTo() method and implement the comparison rules in that method.
The second implementation approach is to customize a comparator object, which needs to inherit the WritableComparator class and override its compare method. Note that in the constructor, the parent class should be called to instantiate the current object participating in comparison. Note that the current object participating in comparison must implement the WritableComparable interface. Finally, set the custom comparator object into the Job in the Job submission code.
11. Describe the rules for implementing sorting comparison in Hadoop (from a source code perspective)
Sorting comparison in Hadoop is essentially about obtaining a comparator object for a certain object. As for the comparison logic, it can be implemented by directly calling the compareTo() method in that comparator object! Next, let's mainly discuss how to obtain a comparator object for an object. From a source code perspective, in the init method of Hadoop's MapTask class, we focus on one line of code: comparator=job.getOutputKeyComparator(); This code is for obtaining the comparator object. Tracing into this method, the source code first checks whether a custom comparator object has been specified in the Job configuration. If the Class file of the already-set comparator object is obtained, the comparator object is then created using reflection. At this point, the process of obtaining the comparator object ends.
If we have not set a custom comparator object in the Job, we cannot obtain the comparator object through reflection. Next, the Hadoop framework will help us create one. The specific approach is as follows:
-
Before obtaining, there is a prerequisite: determine whether the current job's MapOutputKeyClass implements the WritableComparable interface, because we compare based on the key, so we focus on MapOutputKeyClass.
-
If the above step is normal and MapOutputKeyClass implements the WritableComparable interface, next, considering that the objects participating in comparison are Hadoop's own data types, such as Text, LongWritable, etc., these data types have already obtained their comparator objects during class loading, and they are maintained in memory in a HashMap called comparators, with the current object's class as the key and the current object's comparator object as the value. So from the source code, a line of code like WritableComparator comparator = comparators.get(c); is executed, meaning to get the comparator object from the comparators HashMap based on the current object's class file. If nothing unexpected happens, the comparison object for Hadoop's own data type objects can be obtained at this point. However, considering some abnormal situations, such as memory overflow causing GC garbage collection, the comparator object might not be obtainable at this time. In that case, a method called forceInit(c); is executed next to reload the class once more to ensure everything is foolproof. If the comparator object still cannot be obtained at this point, there is only one possibility: the current object participating in comparison is not Hadoop's own data type but our custom object. In the source code, we can also see that a final line of code is executed: comparator = new WritableComparator(c, conf, true);, meaning Hadoop will create a comparator object for us. The above is the entire process of obtaining a comparator object in Hadoop!
12. When writing MR, under what circumstances is Combiner used, and what is the specific implementation process?
The Combiner process is an optional process in MR and is usually an optimization technique. When the data volume after completing the Map phase calculation is relatively large, with too many kv combinations, executing the Reduce phase would cause copying a large amount of data and aggregating more data. To reduce the pressure on Reduce, you can choose to perform Combiner operations in the Map phase, doing some aggregation work in advance. This reduces the number of kv pairs, thereby greatly reducing IO consumption during data transmission.
The general process for implementing Combiner is: first, customize a Combiner class, then inherit Hadoop's Reducer class, override the reduce() method, and perform merge aggregation in that method. Finally, set the custom Combiner class into the Job!
13. Describe the custom implementation process of OutputFormat
The OutputFormat class is the last process in MR. It is mainly responsible for writing out the final data results. Generally, we don't need to customize it; the default is sufficient. However, if we have personalized requirements for the name of the final output result file or the output path, we can implement it through custom OutputFormat. The implementation process is roughly as follows:
First, customize an OutputFormat class, then inherit OutputFormat, override OutputFormat's getRecordWriter() method, and return a RecordWriter object in that method. Since RecordWriter is also an internal Hadoop object, if we want to implement our own logic, we also need to customize a RecordWriter class, then inherit the RecordWriter class, override the write() method and close() method in that class, implement the data writing logic in the write() method, and close resources in the close() method.
14. What is the approach for implementing ReduceJoin in MR, and what are the shortcomings of the ReduceJoin approach?
Speaking of ReduceJoin in MR, first in the Map phase, we uniformly collect data from the two files that need to be joined, manage them with one object, and also add a new attribute to that object to record the source of each piece of data. After collecting the data in the Map phase, write it out directly. When writing out, be sure to pay attention to the choice of the output key. This key must be the join field of the two files. Next, the Reduce phase begins execution. First, copy the data processed by the Map side, then a group of values with the same key enters the reduce() method. Since the key has been previously defined as the join field of the two files, the join for the data entering reduce() this time can be done. The first step is to maintain the data from the two files separately using containers or objects based on different data sources, then traverse one of the containers to obtain the data to be joined according to the specific business logic, and then output the result.
The above is the general approach for ReduceJoin. However, ReduceJoin is relatively performance-intensive, and if data skew scenarios occur, it is even more difficult to handle.
15. What is the approach for implementing MapJoin in MR, and what are the limitations of MapJoin?
MapJoin, as the name suggests, performs the Join directly on the Map side without going through the Reduce phase. This greatly improves the execution efficiency of MR and also solves the problems that data skew brings to the Reduce phase. Next, let's discuss the core approach of MapJoin. First, the prerequisite for MapJoin is that of the two files we need to join, one is a large file and the other is a small file. Under this premise, we can cache the small file in memory in advance, then let the Map side directly process the large file. For each line of data processed, obtain the desired data from memory based on the current join field, and then write out the result.
Comments(0)