数据处理：格式转换与去重的关键策略

### 数据处理：格式转换与去重的关键策略 在数据处理的领域中，高效地组织和处理数据是至关重要的。数据处理管道通常分为通用数据处理管道和自定义业务逻辑管道。下面将重点介绍常见的数据处理步骤，包括文件格式转换和数据去重。 #### 1. 文件格式转换 数据可能以不同的格式进入平台，如 CSV、JSON、XML 文件或自定义二进制格式。传统的数据湖方法是直接存储各种格式的数据，但这种方式在处理多个管道时会面临诸多问题。例如，每个管道都需要单独处理不同的数据格式，当文件格式变化或新增列时，需要更新和测试大量代码，同时也会使数据探索变得复杂。 现代数据平台设计采用更有组织和结构化的方法，将所有传入数据的第一个转换步骤是将其转换为统一的文件格式。我们将使用 Apache Avro 用于暂存区，Apache Parquet 用于生产区。 ##### 1.1 Avro 和 Parquet 文件格式 Avro 和 Parquet 都是二进制文件格式，与 CSV、JSON 和 XML 等文本格式不同，它们不是以人类可读的格式存储，需要特殊程序来解码和编码实际数据。 与基于文本的文件格式相比，二进制文件格式具有以下优点： - **节省磁盘空间**：二进制格式在数据编码过程中可以应用不同的优化，显著减少磁盘占用空间。Avro 和 Parquet 都包含列类型信息，允许更好的文件压缩，从基于文本的文件格式转换为压缩二进制格式，数据大小可减少多达 10 倍。较小的文件大小不仅降低了云存储成本，还显著加快了数据处理管道的速度。 - **强制使用特定模式**：在将数据保存为 Avro 或 Parquet 格式之前，必须定义数据集中存在的列和列类型。在 Avro 文件格式中，模式嵌入到每个文件中，任何读取这些文件的程序或数据管道都将自动了解所有列名及其类型。 为了理解为什么需要同时使用 Avro 和 Parquet 格式，我们需要讨论行导向和列导向文件格式的区别。 行导向文件格式将单个数据行的所有信息保存到连续的文件块中，如 CSV 格式。这种格式在读取文件中所有行的所有列并对其执行操作时非常高效。 列导向文件格式将单个列的值依次存储，即使它们属于不同的行。这种格式在分析工作负载中表现更好，因为只需要某些列来回答问题时，不需要读取整个数据集。此外，列导向格式通常可以获得更好的压缩比。 Avro 是行导向文件格式，支持原始和复杂数据类型，包括嵌套类型。它还支持模式演化规则，适用于暂存区，作为下游转换或临时数据探索用例的数据源。 Parquet 是列导向文件格式，支持原始和复杂数据类型，提供对数据集中单个列的快速访问，显著提高分析查询的性能。它压缩效果好，并且三大主要云仓库（AWS Redshift、Google BigQuery 和 Azure SQL Data Warehouse）都原生支持 Parquet，便于将生产区的数据加载到仓库中。 ##### 1.2 使用 Spark 进行文件格式转换 使用 Apache Spark 作为分布式数据处理框架，将文件从原始格式转换为 Avro 和 Parquet 非常简单。 要在 Spark 中使用 Avro 文件格式，需要一个外部 Avro 库（https://github.com/databricks/spark - avro）。Google Cloud Dataproc 和 Azure Databricks 服务都有预安装的库版本，而对于 AWS EMR 服务，需要在集群创建时明确指定外部库。 如果使用 Spark 2.4.0 或更高版本，则无需外部 Avro 库，因为 Spark 本身已添加了对 Avro 的支持。 以下是一个使用 Spark 将 JSON 文件转换为 Avro 格式的示例代码： ```python import datetime from pyspark.sql import SparkSession spark = SparkSession.builder ... # 为简洁起见，省略 Spark 会话创建 namespace = “ETL” pipeline_name = “click_stream_ingest” source_name = “clicks” batch_id = “01DH3XE2MHJBG6ZF4QKK6RF2Q9” current_date = datetime.datetime.now() in_path = f“gs://landing/{namespace}/{pipeline_name}/{source_name}/{batch_id}/*” out_path = f”gs://staging/{namespace}/{pipeline_name}/{source_name}/year= ➥ {current_date.year}/month={current_date.month}/day={current_date.day}/ ➥ {batch_id}” clicks_df = spark.read.json(in_path) clicks_df = spark.write.format(“avro”).save(out_path) ``` 在这个示例中，我们首先定义了一些变量来组成 Google Cloud Storage（GCS）上的路径，然后读取 JSON 文件并将其保存为 Avro 格式。 #### 2. 数据去重 数据去重是一个重要的话题，主要涉及两个挑战： - **判断相似条目是否代表同一逻辑实体**：例如，客户数据中的 “John Smith” 和 “Jonathan Smith” 是否指同一个人。这通常需要使用主数据管理（MDM）工具来处理。 - **确保数据集中某些属性的唯一性**：例如，确保支付数据集中没有两个记录具有相同的支付