Hadoop数据处理与组织全解析

### Hadoop数据处理与组织全解析 #### 1. CSV输入输出格式处理 在MapReduce中，我们可以编写用于处理CSV的输入输出格式。Pig的piggybank库中有一个CSVLoader，可将CSV文件加载到元组中，它支持CSV记录中的双引号字段，并将每个项作为字节数组提供。还有一个名为csv - serde的GitHub项目（https://github.com/ogrodnek/csv - serde），它有一个Hive SerDe，可对CSV进行序列化和反序列化，和之前的InputFormat示例一样，它也使用OpenCSV项目来读写CSV。 虽然使用TextInputFormat并在映射器中分割行可能更简单，但如果需要多次这样做，可能会陷入复制粘贴的反模式，因为用于标记化CSV的相同代码可能存在于多个位置。如果代码是为了代码复用而编写的，就可以避免这个问题。 #### 2. 输出提交的重要性 在MapReduce中，输出提交是一个关键概念。大多数输出格式使用FileOutputFormat，它使用FileOutputCommitter进行输出提交。当任务和作业执行时，会在某个时间点开始写入作业输出。任务和作业可能会失败、重启或进行推测执行。为了让OutputFormats正确处理这些情况，MapReduce引入了OutputCommitter的概念。 执行流程如下： 1. 当最初咨询FileOutputFormat输出文件的位置时，它将输出位置的决策委托给FileOutputCommitter。 2. FileOutputCommitter指定输出应放到作业输出目录下的临时目录（<job - output>/_temporary/<task - attempt - id>）。 3. 只有在整个任务完成后，FileOutputCommitter才会收到通知，此时临时输出会被移动到作业输出目录。 4. 当整个作业成功完成时，FileOutputCommitter再次收到通知，此时它会在作业输出目录中创建一个_SUCCESS文件，以帮助下游处理器知道作业已成功完成。 如果数据接收器是HDFS，这种机制非常有效。但当处理非文件数据源（如数据库）时，情况会变得更复杂。如果需要进行幂等写入（即相同操作多次应用不会改变结果），则需要在目标数据存储或OutputFormat的设计中考虑这一点。 下面是一个简单的流程说明： ```mermaid graph LR A[任务开始] --> B[写入临时目录] B --> C{任务完成?} C -- 是 --> D[移动临时输出到作业输出目录] D --> E{作业完成?} E -- 是 --> F[创建_SUCCESS文件] C -- 否 --> B E -- 否 --> B ``` #### 3. 数据组织概述 在确定了要使用的数据格式后，就需要考虑如何在HDFS中组织数据。数据组织是使用Hadoop时最具挑战性的方面之一，会受到多种因素的影响，如是否需要提供SQL访问、用于查找数据的字段以及需要支持的访问时间服务级别协议（SLA）等。同时，要避免大量小文件给HDFS NameNode带来不必要的堆压力，并且要学会处理大型输入数据集。 #### 4. 目录和文件布局 建立一个集群范围的标准来定义数据的组织方式是很有价值的，它可以让数据的位置更容易被发现，也有助于对数据存储的公共区域进行结构化管理。常见的方法是创建一个与组织或功能结构相匹配的层级结构。例如，对于分析团队引入的新数据集，目录结构可以如下： ```plaintext /analytics/tweets/raw/v1/2014/07/14/ ``` 其中包含了组名、数据生成日期、数据层（原始、派生或聚合）、数据源名称和数据版本。 为了支持未来数据格式的迁移，可以在目录结构中加入版本号。如果在向数据消费者传达未来的数据变化时遇到挑战，可以考虑使用HCatalog来向客户端抽象数据格式。 另外，按日期和其他字段进行分区也是很有必要的。Hive早期就使用这种技术来加速查询。