Hi，大家好，我是编程小6，很荣幸遇见你，我把这些年在开发过程中遇到的问题或想法写出来，今天说一说
metabase 开源协议_massflow软件,希望能够帮助你!!!。

全文共6522字，预计学习时长19分钟

前不久，AWS re：Invent, Netflix 开源了一个自主开发的构建和管理数据科学项目的框架——Metaflow。在过去的两年里，他们内部数据科学团队迅速地应用它，使得许多项目能够缩短生产时间。

什么是Metaflow?

Metaflow是创建和执行数据科学工作流的框架，并配备了内置功能：

· 管理计算机资源，

· 执行容器化运行，

· 管理外部依赖，

· 版本、重播和恢复工作流运行

· 客户端 API 检查过去的运行适合笔记本电脑,

· 在本地，比如笔记本电脑，和远程云端执行模式之间切换

在内容寻址的数据存储中，Metaflow 自动对代码、数据和依赖关系进行快照，这通常由 S3支持，但也支持本地文件系统。这样就可以总结工作流，重现过去的结果，并检查关于工作流的任何东西，例如笔记本。—— Ycombinator上的vtuulos

基本上它旨在提高数据科学家的生产力，使他们能够专注于实际的数据科学工作，促进其交付成果的更快生产。

每天都有可能发生的事情

· 协作:帮助另一个数据科学家调试错误。希望可以在自己的笔记本电脑中按原样调出他失败的运行状态。

· 恢复运行: 运行失败或故意停止。修复代码中的错误后，希望可以从工作流失败、或停止的地方重新启动工作流。

· 混合运行:希望在本地运行工作流中的一个步骤，可能是数据加载步骤，因为数据集下载文件夹中；但是希望在云上运行另一个计算密集步骤：模型训练。

· 检查运行元数据: 三位数据科学家一直在调整超参数，以便在同一模型上获得更高的准确性。需要分析他们所有的训练运行情况，并选择性能最好的超参数集。

· 相同软件包的多个版本: 希望能够在项目中使用sklearn库的多个版本：0.20版本用于预处理步骤，0.22版本用于建模。

典型的Meta（工作）流是什么样子的？

概念

Metaflow工作流本质上是DAGs (有向无环图) 。下面的图像对其进行了最好的描述。图中的每个节点代表工作流中的一个处理步骤。

Metaflow在工作流程的每个步骤中执行惯用的 python 代码，在独立容器中与其相应的依赖项打包。

Metaflow架构的这个关键方面能够从 Metaflowsconda 生态系统中注入几乎所有的外部库，无需插件。这也是Metaflow 与其他通用解决方案(如Airflow)的不同之处。

程序

只要满足以下最低条件，每个流都可以写成标准的 python 类:

· 继承来自Metaflow的FlowSpec类。

· 每个表示步骤的函数都被@step标记。

· 每个步骤函数应以指向其后续步骤函数结束。可以使用self.next(self.function_name_here)完成此操作

· 实现start和end功能。

下面是最小三节点流three-node flow的示例:

start → process_message → end

1. from metaflow import FlowSpec, step
2. classLinearFlow(FlowSpec):
3. """
4. A flow to verify you can run a basic metaflow flow.
5. """
6. # Global initializations here
7. @step
8. defstart(self):
9. self.message ='Thanks for reading.'
10. self.next(self.process_message)
11. @step
12. defprocess_message(self):
13. print('the message is: %s'%self.message)
14. self.next(self.end)
15. @step
16. defend(self):
17. print('the message is still: %s'%self.message)
18. if__name__=='__main__':
19. LinearFlow()

metaflow-linear-example.py hosted with ❤ by GitHub

最小三节点流: start→ a → end

安装指示

安装和基本运行

· 安装Metaflow（推荐Python3）: pip3 install metaflow

· 将上面的Snippet 1 中的示例代码放在linear_flow.py文件中。

· 看这个流程的设计: python3 linear_flow.py show

· 运行流程: python3 linear_flow.py run

这里有一些值得注意的事情。Metaflow 创建一个本地数据存储.metaflow，它存储与运行关联的所有运行元数据和数据快照。如果配置了云设置，数据快照将存在于 AWS S3Bucket 中，并且运行元数据将加载到由 RDS (关系数据存储)提供的元数据服务中。后文将看到如何使用客户端 API 轻松地检查这个运行元数据。另一个值得注意的微妙但重要的事情是，每个步骤附加的 pid(process ID）是不同的。回到先前的观点，Metaflow将流的每个步骤分装，并在它们自己的环境中运行它们(只是来回传输数据)

Conda设置（如果计划注入依赖项）

· 下载并安装 Miniconda

· 添加 conda频道: conda config --add channelsconda-forge

· 现在可以在工作流中注入conda依赖项了。后文将展示细节。

实施更现实的工作流

在阅读中会介绍及解释 Metaflow 的新概念。

任务

在这个流程中，将实现一个工作流

1. 将CSV插入到Pandas数据框架中

2. 并行计算各种类型的四分位统计数据

3. 保存特定类型统计数据的字典。

流程

下面的框架能帮你更好理解一般流程。

1. from metaflow import FlowSpec, step, catch, retry, IncludeFile, Parameter
2. classGenreStatsFlow(FlowSpec):
3. """
4. A flow to generate some statistics about the movie genres.
5. The flow performs the following steps:
6. 1) Ingests a CSV into a Pandas Dataframe.
7. 2) Compute quartiles for each genre in parallel
8. 3) Save a dictionary of genre specific statistics.
9. """
10. @step
11. defstart(self):
12. """
13. The start step:
14. 1) Loads the movie metadata into pandas dataframe.
15. 2) Finds all the unique genres.
16. 3) Launches parallel statistics computation for each genre.
17. """
18. # TODO: Loading the CSV and getting unique genres
19. self.genres = []
20. self.next(self.compute_statistics, foreach='genres')
21. @catch(var='compute_failed')
22. @retry(times=1)
23. @step
24. defcompute_statistics(self):
25. """Compute statistics for a single genre. Run in cloud"""
26. self.genre =self.input
27. # TODO: Computing statistics for a genre
28. self.next(self.join)
29. @step
30. defjoin(self, inputs):
31. """Join our parallel branches and merge results into a dictionary."""
32. # TODO: Joining the results
33. self.next(self.end)
34. @step
35. defend(self):
36. """End the flow."""
37. pass
38. if__name__=='__main__':
39. GenreStatsFlow()

metaflow_genre_stats_workflow_skeleton.py hosted with ❤ by GitHub

上面的框架中引入了一些概念

· 注意到start步骤第24行的foreach参数了吗？ foreach在 for 循环中为类型列表中的每个条目执行compute_statistics 步骤的并行副本。

· 在26行，@catch(var='compute_failed') 会捕获compute_statistics步骤中发生的任何异常，将其标记为compute_failed 变量，可供后继查看。

· 在27行，@retry(times=1) 做它所暗示的事情，重试步骤，以防出现任何错误。

· 在compute_statistics31行, self.input 是怎么魔幻登场的? input 是Metaflow提出的类变量，包含适用compute_statistics的示例数据。比如当有多个副本的函数并行运行时该怎么做？只有当一个节点分支到多个并行进程时，或者当多个节点合并为一个时，Metaflow才会添加它。

· 这个示例任务只展示相同compute_statistics功能的多个并行运行。但是那些好奇的人可能同时启动完全不同、且毫不相关的功能。要想这么做，把24行换成self.next(self.func1, self.function2,self.function3). 当然也必须相应地更新 join 步骤。

这是框架型的视觉图表

读取数据文件和自定义参数

· 下载此 csv 文件，其中包含Metaflow 准备的电影数据。

· 现在我们希望支持动态传入 movie_data 文件路径和max_genres 值作为外部参数到我们的工作流中。Metaflow 允许通过运行命令中附加额外的标志来传递 args。例如 python3tutorial_flow.py run --movie_data=path/to/movies.csv --max_genres=5

· 为了在工作流中读取这些自定义输入，Metaflow提供了IncludeFile和Parameter 对象。选择用IncludeFile还是Parameter 对象分配给类变量来访问传入的参数，这具体取决于读取的是文件还是常规值。

将Conda 注入流

· 完成上面 Conda 设置部分中概述的步骤。

· 将Metaflow 提供的@conda_base添加到流的类中。它期望传入python版本，可以硬编码，也可以像下面这样通过函数提供。

· 现在可以将@conda添加到流中的任何步骤。它期望通过libraries参数传递具有依赖关系的对象。Metaflow将负责在执行步骤之前准备带有这些依赖项的容器。可以在不同的步骤中使用不同版本的包，因为Metaflow 在不同的容器中运行每个步骤。

· 新运行命令: python3tutorial_flow.py --environment=conda run

实施'start’

有些事情需要注意:

· 注意到在步骤函数中pandas导入语句是如何存在的吗？这是因为它只是由 conda 仅仅注入在这个步骤的范围内。

· 然而，这里定义的变量(dataframe & genres )，甚至可以通过之后运行的步骤来访问。这是因为Metaflow 的工作原理是隔离运行环境，但是允许数据自然地流动。

实现 'compute_statistics’ 步骤

请注意，此步骤正在访问和修改前一个start步骤中定义的dataframe变量。接下来的步骤里，这个新的修改后的dataframe将是有效的。

实施 ‘join’ 步骤

需要关注的事情：

· 在这个步骤中，使用了完全不同版本的pandas库。

· 输入数组中的每个索引都表示在此之前运行的compute_statistics的副本。它包含运行的状态，即各个变量的值。因此，input[0].quartiles可以包含喜剧风格的四分位数，而input[1].quartiles可以包含科幻风格的四分位数。

最终展示：

看它的流程设计：

1. python3 tutorial_flow.py--environment=conda show

运行：

1. python3 tutorial_flow.py--environment=conda run --movie_data=path/to/movies.csv --max_genres=7

通过客户端API检查运行情况

可以利用Metaflow提供的客户端API来检查过去运行的数据和状态快照。它是在笔记本上探索历史运行细节的理想工具。

下面是一个简单的片段，录入了GenreStatsFlow上次成功运行的genre_stats变量。

1. from metaflow import Flow, get_metadata
2. # Print metadata provider
3. print("Using metadata provider: %s"% get_metadata())
4. # Load the analysis from the MovieStatsFlow.
5. run = Flow('GenreStatsFlow').latest_successful_run
6. print("Using analysis from '%s'"%str(run))
7. genre_stats = run.data.genre_stats
8. print(genre_stats)

metaflow_client_api_demo.py hosted with ❤ by GitHub

Metaflow客户端API演示

云端运行

在本地系统上做了一些原型之后，如何利用云资源来加快速度？

目前，Metaflow只支持与AWS的集成，下面是AWS服务Metaflow使用的并排比较。

· 首先，需要在 AWS 上进行一次性设置，以便为 Metaflow 运行创建资源。如果希望共享和访问彼此的运行，您的团队可以利用相同的资源。参考本页面的指示说明。Metaflow为设置提供了一个 Cloudformation 模板，因此毫不费时。

· 然后，在本地系统上运行metaflow configure aws，并提供提示的详细信息。这一步很重要，Metaflow 因此可以使用刚刚在 AWS 上设置的数据存储和元存储。

· 现在，要在云端本地系统上运行任何工作流，只需将--with batch 添加到运行命令中。举个例子，python3 sample_flow.py run--with batch. 简直小菜一碟！

· 对于混合运行，即在本地运行几个步骤，得在云上运行几个步骤——在希望在云上运行的工作流中的步骤中添加 @batch。例如:@batch(cpu=1, memory=500)

Metaflow的局限性与未来展望

· 与AWS的紧密集成，尽管路线图中有支持更多云提供商的计划。

· Metaflow 完全基于CLI，不附带GUI。(不像其他通用的工作流框架，如Airflow)

问题与反馈

如果你有任何进一步的问题或反馈，或如果有任何新的技术分享，请在评论区留言，大家一起讨论~

留言点赞关注

我们一起分享AI学习与发展的干货

如转载，请后台留言，遵守转载规范

今天的分享到此就结束了，感谢您的阅读，如果确实帮到您，您可以动动手指转发给其他人。