简介

数据湖是一个包含文件及其元数据记录的归档。data lake项目包括以下组件

• 监听新文件推送到数据湖并摄取其元数据的摄取器。

• 查询数据湖中文件的API。

• 客户端，它是data lake的Python和命令行界面。您可以使用它将文件推送到数据湖，列出数据湖中可用的文件，并从数据湖检索文件。

要使用此客户端，您（或代表您的人）必须运行data lake-ingester和数据湖-api的一个实例。您将需要从它们那里获取一些配置信息。

为什么我会使用这个？因为您只想把所有文件集中在一个地方，并且具有优雅统一的元数据，这样您可以知道是什么。然后您可以将文件拉到您的硬盘上，享受grep和awk的乐趣。或者，也许您可以将其提供给某种计算集群进行映射和减少。或者，也许您不想设置和维护大量的日志摄取基础设施，或者您不相信该日志摄取基础设施是您事实的来源。或者，也许您只是觉得当事情存档在某处时会感到一种温暖舒适的感觉。

客户端使用

安装

    pip install datalake

如果您计划使用队列功能，您必须安装一些额外的依赖项

    apt-get install libffi-dev # or equivalent
    pip install datalake[queuable]

配置

datalake需要一些配置。每个配置变量都可以在/etc/datalake.conf中设置，作为环境变量设置，或作为参数传递。有关配置变量的文档，请调用datalake --help。

使用方法

datalake具有Python API和命令行客户端。您可以使用一种方法完成的事情，也可以使用另一种方法完成。这是它的工作原理

推送日志文件

    datalake push --start 2015-03-20T00:05:32.345Z
        --end 2015-03-20T23:59.114Z \
        --where webserver01 --what nginx /path/to/nginx.log

使用特定的work-id推送日志文件

    datalake push --start 2015-03-20T00:00:05:32.345Z \
        --end 2015-03-20T00:00:34.114Z \
        --what blappo-etl --where backend01 \
        --work-id blappo-14321359

work-id方便跟踪处理作业或其他实体，这些实体在生命过程中可能通过许多生成日志的机器。它必须在datalake中是唯一的。因此，通常建议在此处使用某种特定领域的前缀。

从指定的起始日期列出webserver01可用的syslog和foobar文件。

    datalake list --where webserver01 --start 2015-03-20 --end `date -u` \
        --what syslog,foobar

使用work id blappo-14321359获取blappo收集、etl和清理日志文件。

    datalake fetch --what gather,etl,cleanup --work-id blappo-14321359

开发者设置

    make docker test

datalake元数据

发送到datalake的文件都附带了JSON元数据文档。如下所示

    {
        "version": 0,
        "start": 1426809920345,
        "end": 1426895999114,
        "path": "/var/log/syslog.1"
        "work_id": null,
        "where": "webserver02",
        "what": "syslog",
        "id": "6309e115c2914d0f8622973422626954",
        "hash": "a3e75ee4f45f676422e038f2c116d000"
    }

version：这是元数据版本。它应该是0。

start：这是文件中第一个事件的时刻，自纪元以来的毫秒数。如果文件与某个瞬间相关联，则此为唯一相关的时间。它是必需的。

end：这是文件中最后一个事件的时刻，自纪元以来的毫秒数。如果键不存在或值为None，则文件表示像周报之类的快照，其中只有一个日期（start）是相关的。

path：原始文件系统中文件的绝对路径。

where：这是生成文件的地理位置或服务器。它是必需的，并且只能包含小写字母数字字符、-和_。它应该简短。'localhost'和'vagrant'是坏名字。像'whirlyweb02-prod'这样的名称是好的。

what：这是生成文件的进程或程序。它是必需的，并且只能包含小写字母数字字符、-和_。它不能有尾部文件扩展名（例如，.log）。名称应该简短，以最大限度地减少与datalake中其他'what'的冲突。所以像'job'或'task'这样的名称是坏的。像'balyhoo-source-audit'或'rawfood-ingester'这样的名称是好的。

id：由datalake分配的文件ID。它是必需的。

hash：文件内容的16字节blake2哈希值。这是由datalake计算和分配的。它是必需的。

work_id：这是一个应用程序特定的ID，以后可以用来检索文件。它是必需的，但可以是null。事实上，datalake实用程序通常会默认将其设置为null，如果未设置。它不能是字符串"null"。它应该以特定领域的名称前缀开头，以防止与其他work id空间冲突。它只能包含小写字母数字字符、-和_。

索引设计

实际上，元数据存储在DynamoDB中，DynamoDB在定义和查询索引方面有一些严格但简单的规则。我们希望支持对元数据的一些简单查询

1. 给我一个给定WHERE的WHATS，从t=START到t=END
2. 从t=START到t=END，给我所有的WHATs
3. 给我给定WHERE和给定WORK_ID的所有WHATs
4. 给我给定WORK_ID的所有WHATs

为了使用DynamoDB实现这一功能，我们采用了“时间桶”的概念，每个时间桶长度为一天。因此，数据跨度从今天1:00-2:00的文件将在今天的时间桶中有一个记录。数据跨度从昨天中午到今天中午的文件将在昨天和今天的时间桶中各有两个记录。因此，当用户查询某个时间段时，我们只需计算该时间段跨越的时间桶，然后在每个时间桶中查找相关文件。

但这不是意味着有时我们必须为每个文件写入多个记录吗？是的。如果一个文件跨越100天怎么办？我们真的想在100个桶中每个都放一个记录吗？嗯，在我们设想的使用场景中，这种情况相当不常见。在实践中，这些文件应该被分割成更小的文件，并更频繁地上传。如果用户查询100天的数据呢？嗯，我们会检查很多桶，这需要一些时间。那些不准备等待这么长时间的用户应该提出更小的请求。

为了实现这些查询，我们有两个哈希和范围索引。它们具有以下哈希键范围键格式

    TIME_BUCKET:WHAT WHERE:ID

    WORK_ID:WHAT WHERE:ID

第一个索引支持查询类型1和2。通过使用TIME_BUCKET:WHAT作为哈希键，我们通过跨WHAT分配写入和查询来防止“热”哈希键。因此，虽然一天内的所有记录都将写入同一个TIME_BUCKET，并且用户更有可能查询最近几个TIME_BUCKETs中的内容，但我们通过多样化的WHAT来分散负载。WHERE:ID范围键可以用于在必要时检索WHEREs的子集。最后，我们追加文件ID以确保键符合DynamoDB的要求。

第二个索引支持查询类型3和4，其模式类似于第一个索引。然而，需要注意的是，WORK_ID是可选的元数据，但在索引用途中是必需的。为了解决这个问题而不在第二个索引中引入热哈希键，摄入器生成一个带有保留前缀“null”的随机WORK_ID。

数据湖记录格式

数据湖客户端指定了当文件推送到数据湖时记录的元数据。我们需要存储一些管理字段，以便我们的查询与DynamoDB索引一起工作。这些记录具有以下格式

    {
        "version": 0,
        "url": "s3://datalake/d-nebraska/nginx/1437375600000/91dd2525a5924c6c972e3d67fee8cda9-nginx-523.txt",
        "time_index_key": "16636:nginx",
        "work_id_index_key": "nullc177bfc032c548ba9e056c8e8672dba8:nginx",
        "range_key": "nebraska:91dd2525a5924c6c972e3d67fee8cda9",
        "create_time": 1426896791333,
        "size": 7892341,
        "metadata": { ... },
    }

version：数据湖记录格式的版本。这里所描述的是版本0。

url：数据湖记录相关资源的URL。

time_index_key：用于基于时间查询的索引的哈希键。它由“时间桶”编号和元数据中的“what”连接而成。

work_id_index_key：用于基于WORK_ID查询的索引的哈希键。它由work_id和元数据中的“what”连接而成。注意，如果work_id为null，将生成一个随机work_id以防止摄入失败和热哈希键。当然，在这种情况下，通过work_id检索是没有意义或不可能的。

range_key：由基于时间和基于WORK_ID的索引使用的范围键。它由元数据中的“where”和“id”连接而成。

create_time：数据湖中文件的创建时间

size：文件的字节数

摄入器

数据湖摄入器将数据湖元数据记录摄入数据库，以便其他数据湖组件可以查询。

摄入器看起来像这样

                                           +----------+     +---------+
       +-------+    +-----------------+    |          |---->| storage |
    -->| queue |--->| s3_notification |--->| ingester |     +---------+
       +-------+    +-----------------+    |          |--+
                                           +----------+  |  +----------+
                                                         +->| reporter |
                                                            +----------+

一个队列接收到数据湖S3存储桶发生事件的通知。一个s3_notification对象将事件从队列的格式转换为数据湖记录格式（见上文）。接下来，摄入器更新存储（即，DynamoDB）并向报告器（即，SNS）报告摄入状态。

数据湖摄入器报告格式

数据湖摄取器为每个摄取的文件生成一个数据湖摄取器报告。报告的格式如下：

    {
        "version": 0,
        "status": "success",
        "start": 1437375854967,
        "duration": 0.738383,
		"records": [
            {
                "url": "s3://datalake/d-nebraska/nginx/1437375600000/91dd2525a5924c6c972e3d67fee8cda9-nginx-523.txt",
                "metadata": { ... }
            }
        ]
    }

版本：数据湖摄取器报告格式的版本。这里所描述的是版本 0。

状态：根据摄取成功与否，可以是“成功”、“警告”或“错误”。如果状态不是“成功”，则“消息”将被设置为可读的说明。

开始：摄取开始时自纪元以来的毫秒数。

持续时间：摄取记录所需的时间（秒）。

记录：已摄取的记录列表。注意，这通常是一个只有一个元素的列表。然而，一些底层协议（例如，s3通知）可能携带关于多个记录的信息。在这些情况下，可能会出现多个记录。

API

数据湖-api为数据湖提供了一个HTTP接口。