构建一个可扩展、可靠和高性能的机器学习（ML）基础架构并不容易。这比用Python构建一个分析模型要花费更多的精力。

Uber已经为许多生产中的用例运行了其可扩展和不依赖框架的机器学习平台Michelangelo，并写了一个很好的总结：

Michelangelo建立初始，最紧急和影响最高的用例是一些非常高规模的问题，这导致我们围绕Apache Spark（用于大规模数据处理和模型训练）和Java（低延迟、高吞吐量的在线服务）构建。这种结构在生产培训和模型部署方面都表现地很好，但是在开销、灵活性和易用性方面，特别是在早期的原型设计和实验中，仍有很多需要改进的地方（这就是Notebook和Python的闪光点）。

Uber将Michelangelo扩展为“对来自任何来源为任何类型的Python模型提供服务，以支持其他机器学习和深度学习框架，如PyTorch 和 TensorFlow （而不是只使用spark进行所有操作）”。

那么，为什么Uber（以及许多其他技术公司）构建了自己的平台和不依赖框架的机器学习基础设施呢？

这篇文章关注的是 Kafka生态系统如何帮助解决数据科学家、数据工程师和生产工程师之间的阻抗失配问题。通过利用它构建自己的可扩展机器学习基础设施并让数据科学家满意，你可以解决类似Uber构建自己的ML平台Michelangelo时所面临的相同问题。

数据科学家、数据工程师和生产工程师之间的阻抗失配

数据科学家、数据工程师和生产工程师之间的阻抗失配是造成公司难以将分析模型投入生产以增加商业价值的主要原因。

下图说明了作为机器学习生命周期中阻抗失配的一部分所需的不同步骤和相应角色：

模型开发和模型部署之间的阻抗失配

数据科学家喜欢Python，仅此而已。因此，大多数机器学习/深度学习框架都集中于Python API。最稳定和最前沿的API以及大多数示例和教程都使用Python API。除了支持Python之外，通常也支持其他编程语言，包括用于Web集成的JavaScript和用于平台集成的Java，尽管这些语言通常具有更少的特性、更不成熟。无论支持其他什么平台，数据科学家都很有可能使用Python构建和训练他们的分析模型。

使用python的模型开发及其工具栈和一个可扩展的、可靠的数据平台之间存在阻抗失配问题，这种平台具有低延迟、高吞吐量、零数据丢失和24/7的可用性要求，这些要求是数据接收、预处理、模型部署和大规模监控所必需的。实际上，对于这些需求，Python并不是最著名的技术。然而，对于像Apache Kafka这样的数据平台来说，它是一个很好的技术。

问题是，编写机器学习源代码以使用Python训练分析模型，你选择的机器学习框架只是现实机器学习基础架构的一小部分。你需要考虑整个模型生命周期。下图显示了机器学习系统中隐藏的技术要务（显示“ML代码”部分有多小）：

因此，你需要对构建到可扩展生产环境中的模型进行训练和部署，以便可靠地利用它。这可以是围绕 Kafka生态系统本地构建的，也可以使用 Kafka只是为了摄取到另一个存储和处理集群中，如HDF或带Spark的AWS S3。Kafka、Spark和其他几个可扩展的基础设施之间存在许多权衡，但这一讨论超出了本文的范围。现在，我们将关注 Kafka。

行业中的不同解决方案可以解决数据科学家、数据工程师和生产工程师之间的某些阻抗失配问题。让我们来看看其中一些选项：

• 官方标准，如开放式神经网络交换（ONNX）、便携式分析格式（PFA）或预测模型标记语言（PMML）：数据科学家用python构建模型。Java开发人员将其导入Java进行生产部署。这种方法支持不同的框架、产品和云服务。你不必依赖相同的框架或产品来进行训练和模型部署。考虑一下ONNX，一个相对较新的深度学习标准——它已经支持TensorFlow、PyTorch和MXNet。这些标准有利弊。有些人喜欢并使用它们；许多人不喜欢。

• 开发人员关注的框架如Deeplearning4j：这些框架是为软件工程师构建的，以便在Java平台上构建整个机器学习生命周期，而不仅仅是模型部署和监控，而且也包括预处理和训练。如果需要，你仍然可以导入其他模型（例如，Deeplarning4J允许你导入Keras模型）。这个选项很好，如果你有可以掌握Java或b的数据科学家或者有足够了解机器学习概念并能构建分析模型的软件工程师。

• 用于构建具有有限机器学习经验的分析模型的AutoML：这样，领域专家可以通过点击按钮来构建和部署分析模型。AutoML引擎为其他人提供了一个界面来使用模型进行预测。

• 将模型二进制文件嵌入到应用程序中：模型训练的输出是一个分析模型。例如，可以编写python代码来训练和生成TensorFlow模型。根据框架，输出可以是文本文件、Java源代码或二进制文件。例如，TensorFlow以Protobuf, JSON和其他文件的形式生成一个模型工件。无论你的机器学习框架的输出是什么格式，它都可以嵌入到应用程序中，以便通过框架的API进行预测（例如，你可以通过TensorFlow的Java API从Java应用程序加载TensorFlow模型）。

• 公共云中的托管模型服务器，如谷歌云机器学习引擎：云提供商承担可用性和可靠性的责任。数据科学家“只是”部署其经过训练的模型，生产工程师可以访问它。关键的权衡是这需要RPC通信来执行模型推断。

尽管所有这些解决方案都有助于数据科学家、数据工程师和生产工程师更好地合作，但隐藏的要务中存在着潜在的挑战：

· 数据收集（即集成）和预处理需要按规模运行

· 对于持续的构建和集成测试，需要共享和自动化配置

· 服务和监控基础架构需要适合你的整体企业架构和工具栈

那么 Kafka生态系统在这里有什么帮助呢？

Apache Kafka作为解决阻抗失配的关键成分

在许多情况下，最好向专家提供他们喜欢和熟悉的工具。挑战是将不同的工具集结合起来，仍然构建一个集成的系统，以及一个连续的、可扩展的机器学习工作流程。因此，在解决数据科学家和开发人员之间的阻抗失配问题时，Kafka并不是具有竞争力的，而是对所讨论的替代方案的补充。

数据工程师使用kafka作为基础设施，使用python进行集成和预处理语句，构建一个可扩展的集成管道。数据科学家可以使用Python或任何其他首选工具构建他们的模型。生产工程师从数据科学家那里获取分析模型（手动或通过任何自动的、连续的集成设置），并将它们嵌入到他们的Kafka应用程序中，以在生产中部署它。或者，团队一起工作，用Java和一个框架，比如Deeplearning4j构建所有的东西。

任何选项都可以与Apache Kafka很好地匹配。选择你需要的部分，无论是用于数据传输的Kafka核心、用于数据集成的Kafka Connect，还是用于数据预处理的Kafka Streams/KSQL。许多组件可用于模型训练和模型推理。如下图所示，编写一次并在两种情况下都使用：

利用Apache Kafka生态系统搭建机器学习基础设施

实时、大规模地监控整个环境也是 Kafka的常见任务。一个巨大的好处是，你只需构建一次高度可靠和可扩展的管道，但将其用于机器学习基础结构的两个部分。你可以在任何环境中使用：在云中，在on-prem数据中心，或者在物联网设备的边缘。

假设你想用KSQL构建一条从MQTT到Kafka的集成管道以进行数据预处理，并使用Kafka Connect将数据导入到HDFS, AWS S3或GoogleCloud Storage中，在那里进行模型训练。相同的集成管道（或者至少部分集成管道）可以重用来进行模型推理。新的MQTT输入数据可以直接用于进行实时预测。

我们刚刚解释了各种解决 Kafka环境中数据科学家和软件工程师之间阻抗失配的可选择方法。现在，让我们在下一节中讨论一个特定的选项，这一选项对于数据科学家来说可能是最方便的：利用Jupyter Notebook中的kafka和KSQL语句，并将其与TensorFlow和Keras结合来训练神经网络。

数据科学家将Python和Jupyter与可扩展的流体系结构结合起来

数据科学家使用诸如Jupyter Notebooks之类的工具来分析、转换、丰富、过滤和处理数据。然后，使用预处理的数据，来训练基于机器学习/深度学习框架（如Tensorflow）的分析模型。

然而，一些数据科学家甚至不知道软件工程师的“面包和黄油”概念，如Github之类的版本控制系统或Jenkins之类的持续集成工具。

这就提出了一个问题，即如何将数据科学家的Python经验与Apache Kafka作为一个久经沙场、高度可扩展的数据处理和流媒体平台的优势结合起来。

数据科学家和数据工程师用Apache Kafka和KSQL

Kafka提供了可以与Python一起使用的集成选项，比如Confluent的用于Apache Kafka的Python客户端或Confluent REST代理用于HTTP集成。但对于习惯于在模型训练和评估之前快速、对数据进行交互式地分析和预处理的数据科学家来说，这并不是一种真正方便的方法。这里通常使用快速原型。

通过编写简单的类SQL语句进行交互式分析和数据预处理，数据科学家可以查看Kafka事件流，并从其著名且喜爱的python环境（如Jupyter）中实现连续的流处理。

下面的python示例利用开源框架ksql python执行来自Kafka流的交互式查询，它在KSQL的rest接口上添加了一个python层。下面是从Jupyter笔记本中使用KSQL的几行Python代码：

该KSQL的查询结果是一个Python生成器对象，你可以很容易地用其他Python库来处理它。这让人感觉更像是土生土长的Python，类似于Numpy、pandas、scikit-learn和其他广泛使用的Python库。

与使用这些库进行快速原型化类似，你可以使用ksqlpython进行交互式查询和数据预处理。查看KSQL快速入门和KSQL方法，了解如何编写KSQL查询，以便轻松筛选、转换、丰富或聚合数据。当KSQL运行连续查询时，你也可以将其用于交互式分析，如果只想获得特定数量的行，可以使用ANSI SQL中的“LIMIT”关键字。

那有什么大不了的？你知道通过ksql-python库KSQL可以感到Python是本机的，但是为什么要使用KSQL来代替或添加到你所熟知和喜爱的python库来分析和处理数据呢？

关键的区别在于，这些KSQL查询也可以在随后的生产中部署。KSQL为你提供了Kafka的所有特性，比如高可扩展性、可靠性和故障转移处理。在Jupyter Notebook中用于交互式分析和预处理的相同KSQL语句可以扩展到每秒数百万条消息。可允许出错。数据丢失为零，语义仅为一次。这对于将热爱Python的数据科学家与高度可扩展和可靠的生产基础设施结合在一起非常重要和有价值。

只是要明确一点：ksql+python并不是所有数据工程任务的全能者，也不会取代现有的Python工具集。但在数据科学家和数据工程师的工具箱中，这是一个很好的选择，增加了新的可能性，例如随着源数据的变化实时更新输入的信息，或者用新的和改进的版本更新部署的模型。

用PythonKSQL和TensorFlow/Keras进行欺诈检测的JupyterNotebook

现在让我们来看一个使用KSQL和Python组合的详细示例。它包括使用ksql python的高级代码示例，以及来自python机器学习生态系统的其他广泛组件，如Numpy、pandas、TensorFlow和Keras。

用例是信用卡支付的欺诈检测。我们使用来自Kaggle的测试数据集作为训练无监督自动编码器的基础，以检测支付中的异常和潜在欺诈。本例的重点不仅仅是模型训练，而是整个机器学习基础设施，包括数据接收、数据预处理、模型训练、模型部署和监控。所有这些都需要具有可扩展性、可靠性和表现力。

让我们看一下JupyterNotebook的一些片段。

连接到KSQL服务器并使用python创建KSQL流：

from ksql import KSQLAPI
client = KSQLAPI('http://localhost:8088')
client.create_stream(table_name='creditcardfraud_source',
 columns_type=['Id bigint', 'Timestamp varchar', 'User varchar', 'Time int', 'V1 double', 'V2 double', 'V3 double', 'V4 double', 'V5 double', 'V6 double', 'V7 double', 'V8 double', 'V9 double', 'V10 double', 'V11 double', 'V12 double', 'V13 double', 'V14 double', 'V15 double', 'V16 double', 'V17 double', 'V18 double', 'V19 double', 'V20 double', 'V21 double', 'V22 double', 'V23 double', 'V24 double', 'V25 double', 'V26 double', 'V27 double', 'V28 double', 'Amount double', 'Class string'],
 topic='creditcardfraud_source',
 value_format='DELIMITED')

使用python预处理传入的付款信息：

· 过滤不需要的列

· 过滤列“class”为空的消息

· 将数据格式更改为AVRO，以方便进一步处理

client.create_stream_as(table_name='creditcardfraud_preprocessed_avro',
 select_columns=['Time', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6', 'V7', 'V8', 'V9', 'V10', 'V11', 'V12', 'V13', 'V14', 'V15', 'V16', 'V17', 'V18', 'V19', 'V20', 'V21', 'V22', 'V23', 'V24', 'V25', 'V26', 'V27', 'V28', 'Amount', 'Class'],
 src_table='creditcardfraud_source',
 conditions='ClassIS NOT NULL',
 kafka_topic='creditcardfraud_preprocessed_avro',
 value_format='AVRO')

使用KSQL进行可能的数据争用和预处理的更多示例：

· 删除列，过滤值“class”为空的消息，并将数据格式更改为Avro：

CREATE STREAM creditcardfraud_preprocessed_avro WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='creditcardfraud_preprocessed_avro') ASSELECTTime, V1 ,V2 , V3 , V4 , V5 , V6 , V7 , V8 , V9 , V10 , V11 , V12 , V13 , V14 , V15 , V16, V17 , V18 , V19 , V20 , V21 , V22 , V23 , V24 , V25 , V26 , V27 , V28 ,Amount , Class FROM creditcardfraud_source WHERE Class ISNOTNULL;

· 匿名（屏蔽最左边的两个字符，例如，“Hans”变为“**ns”）：

SELECT Id, MASK_LEFT(User, 2) FROM creditcardfraud_source;

· 增强（如果“class”为空，则添加-1）:

SELECT Id, IFNULL(Class, -1) FROM creditcardfraud_source;

· 合并/连接数据帧:

CREATE STREAM creditcardfraud_per_user WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='creditcardfraud_preprocessed_avro') ASSELECTTime, V1 ,V2 , V3 , V4 , V5 , V6 , V7 , V8 , V9 , V10 , V11 , V12 , V13 , V14 , V15 , V16, V17 , V18 , V19 , V20 , V21 , V22 , V23 , V24 , V25 , V26 , V27 , V28 ,Amount , Class FROM creditcardfraud_enahnced c INNER JOIN USERS u on c.userid = u.useridWHERE V1 > 5AND V2 ISNOTNULLAND u.CITYLIKE'Premium%';

Jupyter Notebook包含完整的示例。我们使用Python+KSQL进行集成、数据预处理和交互分析，并将它们与通用的Python机器学习工具栈中的各种其他库结合起来进行原型设计和模型训练：

· 使用Numpy和pandas处理数组/矩阵

· 使用scikit-learn进行 ML特定处理（拆分训练/测试等）

· 通过Matplotlib数据可视化进行交互分析

· 使用TensorFlow和Keras进行ML训练+评估

模型推理和可视化也在Jupyternotebook中完成。在构建了一个准确的模型之后，你可以将其部署到任何地方进行预测，并利用相同的集成管道进行模型训练。在Kafka环境中部署模型的一些示例包括：

· 嵌入在Kafka Streams微服务中的分析模型（TensorFlow、Keras、H2O和Deeplarning4j）

· 利用嵌入到KSQL UDF中的模型对物联网传感器数据进行异常检测

· Kafka Streams应用程序和模型服务器之间的RPC通信（TensorFlow服务）

用于原型、演示和生产部署的Python、KSQL和Jupyter

诚如所见，无论在理论上（谷歌的论文-机器学习系统中的隐藏技术要务）还是在实践中（Uber的机器学习平台Michelangelo），构建一个可扩展、可靠和性能良好的机器学习基础设施都不是一个简单的任务。

必须解决数据科学家、数据工程师和生产工程师之间的阻抗失配问题，才能使机器学习项目实现真正的业务价值。这需要使用该工作所需的工具，并了解如何对其进行组合。你可以使用Python和Jupyter进行原型制作和演示（如果你只想在历史数据集上进行快速、简单的原型制作，通常Kafka和ksql会在这里开销很大，也不需要这样做），或者将Python和Jupyter与你的整个开发生命周期结合起来，直至大规模的生产部署。

将Kafka事件流和KSQL语句集成到Jupyter Notebook中可以：

· 使用数据科学家（包括Python和Jupyter）首选的现有环境，并将其与Kafka和KSQL结合，通过使用简单的Python包装器API执行KSQL查询，集成并持续处理实时流式数据。

· 轻松连接实时流数据，而不仅仅是历史数据批次（可能是从前一天、一周或一个月开始，例如通过CSV文件进入）

· 将来自Kafka的基于流式事件的传感器数据等不同概念与诸如生成器或字典对象之类的Python编程概念结合起来，这些概念可用于你的Python数据处理工具或诸如Numpy、pandas或scikit-learn之类的ML框架。

· 重复使用相同的逻辑进行集成、预处理和监控，并将其从Jupyter Notebook和原型制作或演示转移到大型测试和生产系统中。

用于原型设计的python和用于可扩展流媒体平台的Apache Kafka不是竞争的技术栈。两者可以很好地协同工作，特别是如果你使用诸如jupyterNotebook和KSQLl之类的“助手工具”。