华为云用户手册

安全认证代码（Java版） 目前样例代码统一调用LoginUtil类进行安全认证。 在Spark样例工程代码中，不同的样例工程，使用的认证代码不同，基本安全认证或带ZooKeeper认证。样例工程中使用的示例认证参数如表2所示，请根据实际情况修改对应参数值。 表2 参数描述 参数 示例参数值 描述 userPrincipal sparkuser 用户用于认证的账号Principal，您可以联系管理员获取此账号。 userKeytabPath /opt/FIclient/user.keytab 用户用于认证的Keytab文件，您可以联系管理员获取文件。 krb5ConfPath /opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf krb5.conf文件路径和文件名称。 ZKServerPrincipal zookeeper/hadoop.hadoop.com ZooKeeper服务端principal。请联系管理员获取对应账号。 基本安全认证： Spark Core和Spark SQL程序不需要访问HBase或ZooKeeper，所以使用基本的安全认证代码即可。请在程序中添加如下代码，并根据实际情况设置安全认证相关参数： String userPrincipal = "sparkuser"; String userKeytabPath = "/opt/FIclient/user.keytab"; String krb5ConfPath = "/opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf"; Configuration hadoopConf = new Configuration(); LoginUtil.login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf); 带ZooKeeper认证： 由于“Spark Streaming”、“通过JDBC访问Spark SQL”和“Spark on HBase”样例程序，不仅需要基础安全认证，还需要添加ZooKeeper服务端Principal才能完成安全认证。请在程序中添加如下代码，并根据实际情况设置安全认证相关参数： String userPrincipal = "sparkuser"; String userKeytabPath = "/opt/FIclient/user.keytab"; String krb5ConfPath = "/opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf"; String ZKServerPrincipal = "zookeeper/hadoop.hadoop.com"; String ZOOKEEPER_DEFAULT_ LOG IN_CONTEXT_NAME = "Client"; String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal"; Configuration hadoopConf = new Configuration(); LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userPrincipal, userKeytabPath); LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZKServerPrincipal); LoginUtil.login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf);

场景说明 假定HBase的table1表存储用户当天消费的金额信息，table2表存储用户历史消费的金额信息。 现table1表有记录key=1,cf:cid=100，表示用户1在当天消费金额为100元。 table2表有记录key=1,cf:cid=1000，表示用户1的历史消息记录金额为1000元。 基于某些业务要求，要求开发Spark应用程序实现如下功能： 根据用户名累计用户的历史消费金额，即用户总消费金额=100(用户当天的消费金额) + 1000(用户历史消费金额)。 上例所示，运行结果table2表用户key=1的总消费金融为cf:cid=1100元。

数据规划 使用HBase shell工具，创建HBase table1和table2，并分别插入数据。 通过HBase创建名为table1的表，命令如下。 create 'table1', 'cf' 通过HBase插入数据，命令如下： put 'table1', '1', 'cf:cid', '100' 通过HBase创建名为table2的表，命令如下。 create 'table2', 'cf' 通过HBase插入数据。 put 'table2', '1', 'cf:cid', '1000' 如果开启了kerberos认证，需要将客户端的配置文件“spark-defaults.conf”和sparkJDBC服务端中的配置项spark.yarn.security.credentials.hbase.enabled置为true。

Alluxio简介 Alluxio是一个面向基于云的数据分析和人工智能的开源的数据编排技术。它为数据驱动型应用和存储系统构建了桥梁, 将数据从存储层移动到距离数据驱动型应用更近的位置，从而能够更容易、更快的被访问。同时使得应用程序能够通过一个公共接口连接到许多存储系统。 Alluxio主要特点如下： 提供内存级I/O 吞吐率，同时降低具有弹性扩张特性的数据驱动型应用的成本开销 简化 云存储 和对象存储接入 简化数据管理，提供对多数据源的单点访问 应用程序部署简易

Spark Streaming常用接口 Spark Streaming中常见的类有： pyspark.streaming.StreamingContext：是Spark Streaming功能的主入口，负责提供创建DStreams的方法，入参中需要设置批次的时间间隔。 pyspark.streaming.DStream：是一种代表RDDs连续序列的数据类型，代表连续数据流。 dstream.PariDStreamFunctions：键值对的DStream，常见的操作如groupByKey和reduceByKey。 对应的Spark Streaming的JAVA API是JavaSteamingContext，JavaDStream和JavaPairDStream。 Spark Streaming的常见方法与Spark Core类似，下表罗列了Spark Streaming特有的一些方法。 表3 Spark Streaming常用接口介绍 方法 说明 socketTextStream(hostname, port, storageLevel) 从TCP源主机：端口创建一个输入流。 start() 启动Streaming计算。 awaitTermination(timeout) 当前进程等待终止，如Ctrl+C等。 stop(stopSparkContext, stopGraceFully) 终止Streaming计算,stopSparkContext用于判断是否需要终止相关的SparkContext，StopGracefully用于判断是否需要等待所有接受到的数据处理完成。 UpdateStateByKey(func) 更新DStream的状态。使用此方法，需要定义State和状态更新函数。 window(windowLength, slideInterval) 根据源DStream的窗口批次计算得到一个新的DStream。 countByWindow(windowLength, slideInterval) 返回流中滑动窗口元素的个数。 reduceByWindow(func, windowLength, slideInterval) 当调用在DStream的KV对上，返回一个新的DStream的KV对，其中每个Key的Value根据滑动窗口中批次的reduce函数聚合得到。 join(other,numPartitions) 实现不同的Spark Streaming之间做合并操作。

SparkSQL常用接口 Spark SQL中在Python中重要的类有： pyspark.sql.SQLContext：是Spark SQL功能和DataFrame的主入口。 pyspark.sql.DataFrame：是一个以命名列方式组织的分布式数据集。 pyspark.sql.HiveContext：获取存储在Hive中数据的主入口。 pyspark.sql.DataFrameStatFunctions：统计功能中一些函数。 pyspark.sql.functions：DataFrame中内嵌的函数。 pyspark.sql.Window：sql中提供窗口功能。 表4 Spark SQL常用的Action 方法 说明 collect() 返回一个数组，包含DataFrame的所有列。 count() 返回DataFrame中的行数。 describe() 计算统计信息，包含计数，平均值，标准差，最小值和最大值。 first() 返回第一行。 head(n) 返回前n行。 show() 用表格形式显示DataFrame。 take(num) 返回DataFrame中的前num行。 表5 基本的DataFrame Functions 方法 说明 explain() 打印出SQL语句的逻辑计划和物理计划。 printSchema() 打印schema信息到控制台。 registerTempTable(name) 将DataFrame注册为一张临时表，命名为name，其周期和SQLContext绑定在一起。 toDF() 返回一个列重命名的DataFrame。

部署运行 在Storm示例代码根目录执行如下命令打包："mvn package"。执行成功后，将会在target目录生成storm-examples-1.0.jar。 执行命令提交拓扑。提交命令示例（拓扑名为jdbc-test）。 storm jar /opt/jartarget/storm-examples-1.0.jar com.huawei.storm.example.jdbc.SimpleJDBCTopology jdbc-test

数据库配置—Derby数据库配置过程 首先应下载一个数据库，可根据具体场景选择最适合的数据库。 该任务以Derby数据库为例。Derby是一个小型的，java编写的，易于使用却适合大多数应用程序的开放源码数据库。 Derby数据库的获取。在官网下载最新版的Derby数据库(本示例使用10.14.1.0)，通过WinScp等工具传入Linux客户端，并解压。 在Derby的安装目录下，进入bin目录，输入如下命令。 export DERBY_INSTALL=/opt/db-derby-10.14.1.0-bin export CLASSPATH=$DERBY_INSTALL/lib/derbytools.jar:$DERBY_INSTALL\lib\derbynet.jar:. export DERBY_HOME=/opt/db-derby-10.14.1.0-bin . setNetworkServerCP ./startNetworkServer -h 主机名 执行./ij命令，输入connect 'jdbc:derby://主机名:1527/example;create=true';，建立连接。 执行./ij命令前，需要确保已配置java_home,可通过which java命令检查是否已配置。 数据库建立好后，可以执行sql语句进行操作，需要建立两张表ORIGINAL和GOAL，并向ORIGINAL中插入一组数据，命令如下：（表名仅供参考，可自行设定） CREATE TABLE GOAL(WORD VARCHAR(12),COUNT INT ); CREATE TABLE ORIGINAL(WORD VARCHAR(12),COUNT INT ); INSERT INTO ORIGINAL VALUES('orange',1),('pineapple',1),('banana',1),('watermelon',1);

操作场景 本文档主要说明如何使用开源Storm-JDBC工具包，完成Storm和JDBC之间的交互。Storm-JDBC中包含两类Bolt：JdbcInsertBolt和JdbcLookupBolt。其中，JdbcLookupBolt主要负责从数据库中查数据，JdbcInsertBolt主要向数据库中存数据。当然，JdbcLookupBolt和JdbcInsertBolt中也可以增加处理逻辑对数据进行处理。 本章节只适用与 MRS 产品Storm与JDBC组件间的访问。本章中描述的jar包的具体版本信息请以实际情况为准。

SparkSQL常用接口 Spark SQL中常用的类有： SQLContext：是Spark SQL功能和DataFrame的主入口。 DataFrame：是一个以命名列方式组织的分布式数据集。 HiveContext：获取存储在Hive中数据的主入口。 表5 常用的Actions方法 方法 说明 collect(): Array[Row] 返回一个数组，包含DataFrame的所有列。 count(): Long 返回DataFrame中的行数。 describe(cols: String*): DataFrame 计算统计信息，包含计数，平均值，标准差，最小值和最大值。 first(): Row 返回第一行。 Head(n:Int): Row 返回前n行。 show(numRows: Int, truncate: Boolean): Unit 用表格形式显示DataFrame。 take(n:Int): Array[Row] 返回DataFrame中的前n行。 表6 基本的DataFrame Functions 方法 说明 explain(): Unit 打印出SQL语句的逻辑计划和物理计划。 printSchema(): Unit 打印schema信息到控制台。 registerTempTable(tableName: String): Unit 将DataFrame注册为一张临时表，其周期和SQLContext绑定在一起。 toDF(colNames: String*): DataFrame 返回一个列重命名的DataFrame。

Spark Streaming常用接口 Spark Streaming中常见的类有： StreamingContext：是Spark Streaming功能的主入口，负责提供创建DStreams的方法，入参中需要设置批次的时间间隔。 dstream.DStream：是一种代表RDDs连续序列的数据类型，代表连续数据流。 dstream.PariDStreamFunctions：键值对的DStream，常见的操作如groupByKey和reduceByKey。 对应的Spark Streaming的JAVA API是JavaSteamingContext，JavaDStream和JavaPairDStream。 Spark Streaming的常见方法与Spark Core类似，下表罗列了Spark Streaming特有的一些方法。 表3 Spark Streaming方法介绍 方法 说明 socketTextStream(hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): ReceiverInputDStream[String] 用TCP协议（源主机:端口）创建一个输入流。 start():Unit 启动Streaming计算。 awaitTermination(timeout: long):Unit 当前进程等待终止，如Ctrl+C等。 stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit 终止Streaming计算。 transform[T](dstreams: Seq[DStream[_]], transformFunc: (Seq[RDD[_]], Time) ? RDD[T])(implicit arg0: ClassTag[T]): DStream[T] 对每一个RDD应用function操作得到一个新的DStream。 UpdateStateByKey(func) 更新DStream的状态。使用此方法，需要定义状态和状态更新函数。 window(windowLength, slideInterval) 根据源DStream的窗口批次计算得到一个新的DStream。 countByWindow(windowLength, slideInterval) 返回流中滑动窗口元素的个数。 reduceByWindow(func, windowLength, slideInterval) 当调用在DStream的KV对上，返回一个新的DStream的KV对，其中每个Key的Value根据滑动窗口中批次的reduce函数聚合得到。 join(otherStream, [numTasks]) 实现不同的Spark Streaming之间做合并操作。 DStreamKafkaWriter.writeToKafka() 支持将DStream中的数据批量写入到Kafka。 DStreamKafkaWriter.writeToKafkaBySingle() 支持将DStream中的数据逐条写入到Kafka。 表4 Streaming增强特性接口 方法 说明 DStreamKafkaWriter.writeToKafka() 支持将DStream中的数据批量写入到Kafka。 DStreamKafkaWriter.writeToKafkaBySingle() 支持将DStream中的数据逐条写入到Kafka。

使用场景 场景挑战 设备采集的数据信息量庞大但只需部分有效数据。部署在节点下的设备需要在本地判断情况以及时进行控制。 设备使用的协议不能接入边缘计算。 节点下的IT子系统众多，接入时不便于管理，配置繁琐。子系统采集的数据需要传输到用户的北向应用。 解决方案 通过开发集成ModuleSDK的应用，您可以： 使用ModuleSDK提供的AppClient开发OT应用对需要上传到云端的信息进行过滤上传。OT应用可以在本地对设备进行命令的下发。 使用ModuleSDK提供的DriverClient开发驱动应用进行协议转换。 使用ModuleSDK提供的ItClient开发IT应用实现子系统与云端配置的自动同步，也可以实现云端对子系统的配置管理。云端可以通过IT应用实现对子系统数据的采集。

简介 为解决用户自定义处理设备数据以及自定义协议设备快速接入IOT平台的诉求，IoT边缘提供ModuleSDK，用户可通过集成SDK让设备以及设备数据快速上云。IoT边缘平台应用功能有自定义处理设备数据（即数据处理），自定义协议设备快速接入（即协议解析），IT子系统接入(即IT应用)，并且支持容器化部署和安装包部署的方式。 集成ModuleSDK开发的应用被称为插件应用，其最终将被通过云部署到边缘节点。 集成ModuleSDK的应用分为： OT应用：数据处理类型的应用，实现了总线回调接口，IoT边缘云服务为OT应用提供总线对接能力、设备反向控制（命令）能力。 驱动应用：协议转换类型的应用，可将使用自定义协议的设备接入边缘计算。 IT应用：实现工业子系统接入的应用，实现了对子系统配置同步、反向调用接口、以及数据采集能力。 集成了SDK的应用称为插件，华为云市场提供了插件上架功能。您可以将你开发的集成了SDK的应用上传到华为云市场来获取收益，也可以在华为云市场下载使用第三方插件。

镜像包打包 上传打包的项目。 将jar文件上传到联网的linux机器上，如目录（/home/monitor）中 安装docker。 请确认你使用的系统已经安装docker(docker版本需要高于17.06，推荐18.06),安装参照docker 安装教程 制作镜像。 搜索基础镜像，基础镜像需要集成jre。 docker search jre8 选择合适的镜像（镜像需要集成了jre且版本不低于8） NAME DESCRIPTION STARS OFFICIAL AUTOMATED livingobjects/jre8 Jre8 image 4 [OK] livingobjects/jre8镜像是docker hub第三方提供的镜像，非IoT团队发布，且IoT团队未提供任何官方镜像。该镜像在此仅做示例，IoT团队对该镜像的安全性不作保证。强烈建议用户自己封装镜像！ 拉取镜像 docker pull livingobjects/jre8 编写dockerfile制作镜像 dockerfile内容参照如下（具体可参考编写高效的Dockerfile ） #Version 1.0.0 FROM livingobjects/jre8 #基础镜像来源 RUN mkdir -p /opt/iot/edge/monitor / && chmod -R 777 /opt/ #授权 COPY monitor /opt/iot/edge/monitor #复制文件到指定目录,此dockerfile文件位置:/home,jar包位置:/home/monitor USER root #用户名 EXPOSE 8080 CMD ["java", "-jar", "/opt/iot/edge/monitor/monitor-app.jar", "run"] #运行命令 构建镜像 docker build -t edge_monitor:1.0.0 /home --no-cache 查看打包完成的镜像 docker images REPOSITORY TAG IMAGE ID CREATED SIZE edge_monitor 1.0.0 93f9d964bcea 12 seconds ago 243MB 镜像上传。 上传镜像 开通 容器镜像服务 SWR 镜像上传需要使用SWR镜像容器服务，开通及使用请参照容器 镜像服务 SWR_用户指南 获取SWR登录指令 请参照使用容器引擎客户端上传镜像第四节。 tag镜像 使用tag命令将镜像重命名以确定上传位置和组织，获取上传位置和组织前缀请参考pull/push命令 docker tag edge_monitor:1.0.0 swr.cn-north-4.myhuaweicloud.com/hw_swr/edge_monitor:1.0.0 上传镜像 docker push swr.cn-north-4.myhuaweicloud.com/hw_swr/edge_monitor:1.0.0 查看上传结果 上传镜像后请在SWR将镜像设置为公开。

镜像包打包 上传需要打包的项目。将项目上传到网络能访问到的Linux机器上，放到目录下（比如 /home/MyCApp）。 安装docker。 请确认你使用的系统已经安装docker(docker版本需要高于17.06，推荐18.06)，安装方法可参照docker 安装教程。 制作镜像。 搜索基础镜像。 对基础镜像没有要求，可自行选择合适的基础镜像，以下选用带cmake的镜像作为示例 docker search cmake 选择合适的镜像（镜像需要集成了cmake且版本不低于3.9.5） NAME DESCRIPTION STARS OFFICIAL AUTOMATED lycantropos/cmake CMake Docker image 3 [OK] 拉取镜像 lycantropos/cmake镜像是docker hub第三方提供的镜像，非IoT团队发布，且IoT团队未提供任何官方镜像。该镜像在此仅做示例，IoT团队对该镜像的安全性不作保证。强烈建议用户自己封装镜像！ docker pull lycantropos/cmake 添加启动脚本start.sh，放到项目文件下（和main.c在一个目录下）。 function log(){ echo `date "+%Y-%m-%d %T"`: $1 } log "[INFO] start execute process." # 这里的路径取决于项目保存的位置 cd /opt/iot/edge/MyCApp ./MyCApp MyCApp为可执行文件，生成步骤可参考生成可执行文件 编写 Dockerfile 制作镜像 创建 Dockerfile，命名为 myapp-dockerfile 内容参照如下（具体可参考编写高效的Dockerfile ）。 下面提供myapp-dockerfile 样例，请根据具体需要修改。 注意myapp-dockerfile需要和项目放到一个目录下。 #Version 1.0.0 # 基础镜像来源 # 如果不采用在镜像中编译源文件的方式，可以任意选择基础镜像 FROM lycantropos/cmake # 创建镜像文件目录，并且授权 RUN mkdir -p /opt/iot/edge/MyCApp/conf && chmod -R 777 /opt ENV docker_path=/opt/iot/edge/MyCApp ENV LD_LIBRARY_PATH=${docker_path}:$LD_LIBRARY_PATH WORKDIR ${docker_path} # 复制工程或文件到指定目录 COPY MyCApp/MyCApp $docker_path # 将依赖库和配置文件放到对应目录下，确保编译的时候不会出错 COPY MyCApp/lib /usr/lib COPY MyCApp/conf $docker_path/conf COPY MyCApp/start.sh $docker_path # 指定容器将要监听的端口 USER root EXPOSE 8082 ENTRYPOINT ["/bin/bash", "/opt/iot/edge/MyCApp/start.sh"] 构建镜像 docker build -f ./myapp-dockerfile -t my_app_docker:v1.0.0 ./ 查看打包完成的镜像 docker images 可以看到my_app_docker这个镜像已经制作完成。 REPOSITORY TAG IMAGE ID CREATED SIZE my_app_docker v1.0.0 983b4e5aa72a 10 minutes ago 1.51GB 镜像上传 以上步骤可以通过体验馆熟悉流程。 上传镜像 镜像上传需要使用镜像容器服务(SWR)，首先需要开通容器镜像服务(SWR)。开通及使用请参照容器镜像服务(SWR)。 获取 SWR 登录指令 获取登录指令请参照获取指令。 访问密钥即AK/SK（Access Key ID/Secret Access Key），获取的密钥和AK将用于登录。 登录 SWR 仓库 docker login -u [区域项目名]@[AK] -p [登录密钥] [镜像仓库地址] 可以直接从控制台获取登录命令，如下图。 镜像仓库地址 = swr.区域项目名称.myhuaweicloud.com 例如，华北-北京一对应的镜像仓库地址为：swr.cn-north-1.myhuaweicloud.com 修改镜像所属组织 修改镜像的组织名，以便推送到个人组织内。 docker tag [OPTIONS] [镜像名:版本号] [镜像仓库地址/所属组织/镜像名:版本号] 例如， docker tag my_app_docker:v1.0.0 swr.cn-north-4.myhuaweicloud.com/iotedge/my_app_docker:v1.0.0 上传镜像 docker push [镜像仓库地址/所属组织/镜像名:版本号] 例如， docker push swr.cn-north-4.myhuaweicloud.com/iotedge/my_app_docker:v1.0.0 在我的镜像查看上传结果 上传镜像后请在 SWR 将镜像设置为公开。 步骤一： 步骤二： 步骤三： 这一步很重要，关系到后面能否正常部署应用。

代码解析 代码解释使用ModuleSDK开发应用集成ModuleSDK进行数据处理。 AppClient类有以下几个关键方法（具体参考JavaDoc）。 createFromEnv(): AppClient创建时由此方法自动获取环境变量。 setBusMessageCallback(): 设置总线消息回调，用于对设备上报的数据进行处理 sendBusMessage(): 向总线发送消息，用于将处理后的设备数据发送到总线 callDeviceCommand(): 调用设备命令 getDevicesInfo(): 查询设备状态 MonitorApp代码解析 片段一 privatestaticfinalString INPUT = "input"; public static final String OUTPUT = "output"; 定义输入和输出的端点，关于取值需要需在创建应用版本的inputs参数中定义,创建应用时输入端点与输出端点以及数据流转规则的配置与此是对应的。例如此处定义了输入端点为“input”，输出端点为“output”，则创建应用时的端点和软件配置输入端点需要配置为input，输出端点需要配置为output。 应用部署后还需要设置数据流转规则后，决定数据的流向。 /** * 电机设备的产品ID */ public static final String MOTOR_PRODUCT_ID = "60988d94aa3bcc02c0200667"; 单击设备的产品ID,需要在IoTDA设备接入创建产品时获取。 片段二 private AppClient appClient; publicMonitorApp() throws GeneraException { appClient = AppClient.createFromEnv(); } 定义并创建AppClient， AppClient.createFromEnv()创建其配置参数将会自动从边缘节点环境中获取，数据的传输将会依赖AppClient。 片段三 publicvoidstart() throws GeneraException { //设置回调，打开客户端 appClient.setBusMessageCallback(INPUT, this);//设置收到设备数据的回调 appClient.open(); } appClient在接收到数据后的处理动作需要用户定义，具体操作是设置回调。 这里设置回调方法并传入输入端点后，appClient将会开启并启动一个监听器监听输入端点的数据传输，接收到设备经过hub发来的数据后会调用回调进行数据处理。 片段四 public void onMessageReceived(BusMessage busMessage) { try { if (busMessage.getProductId().equals(MOTOR_PRODUCT_ID)) { //马达设备状态错误时对马达进行重启 MotorData motorData = JsonUtil.fromJson( JsonUtil.toJson(busMessage.getServices().get(0).getProperties()), MotorData.class); if (motorData.getStatus().equals("error")) { Command command = new Command(busMessage.getDeviceId(),"power", "power_control", "restart" ); appClient.callDeviceCommand(command, FIVE_SECOND); } } else { //其他设备数据发布到总线 appClient.sendBusMessage(OUTPUT, busMessage); } } catch (Exception e) { System.out.println(e.getMessage()); } } onMessageReceived(BusMessage busMessage)是前面设置回调具体的回调函数，它是BusMessageCallback接口的方法，MonitorApp要实现BusMessageCallback接口并实现此方法，AppClient通过设置的输入端口input监听到设备发送数据时调用此方法进行数据处理，用户关于设备发送的数据的处理逻辑在此方法内实现，处理后的数据通过设置的输出端口output发送经过hub发送到云端。 appClient.callDeviceCommand(command, FIVE_SECOND)是示例演示应用在接受到设备传来的error信号后，向设备发送秒后重启命令。 父主题： 集成ModuleSDK进行数据处理

安装包部署 软件部署配置 部署方式选择安装包部署，"安装包地址"为{桶名/对象名}。如桶名为:“Demo”，对象名为:“ModuleSDK-Demo.zip”， 则安装包为obs://Demo/ModuleSDK-Demo.zip。 容器规格与高级配置请根据您的需要进行配置。 运行配置 输入端点、输出端点与demo中代码定义的端点一一对应，由于本例演示的是数采模块，因此不对端点进行配置。 部署配置请根据您的需要进行选择。 输入输出端点是非必需配置的，当有数据流转时需要配置，如OT应用（数据处理）。 驱动类应用和IT应用一般不需要配置。 配置确认 建议直接单击“立即发布”，方便后面直接部署应用的时候，能够获取到最新版本。

步骤一：配置监控数据存储路径 Zabbix会将监控数据保存在其所在的机器上，您可以根据如下步骤设置监控数据的存储路径。 登录Zabbix所在服务器。 打开zabbix_server.conf文件。 vim /etc/zabbix/zabbix_server.conf 在zabbix_server.conf文件中，设置数据存储路径。 ExportDir=/tmp/ 重启Zabbix服务，使配置生效。 systemctl restart zabbix-server 配置生效后，Zabbix会在/tmp目录下生产文件（文件名后缀为.ndjson），用于保存监控数据。

步骤四：查看实时日志 完成日志接入配置后，可以在 云日志 服务控制台实时查看上报的日志。 如果您正在使用实时查看功能，请停留在实时查看页面，请勿切换页面。如果您离开实时查看页面，实时查看功能将会被关闭。 在“接入管理”页面，单击目标日志接入任务“所属日志流”列的日志流名称，即可进入日志流详情页。 单击“实时日志”页签，查看实时日志。 日志大约每隔5秒钟上报一次，在日志消息区域，您最多需要等待5秒钟左右，即可查看实时上报的日志。 图10 实时日志

前提条件 注册账号并实名认证。使用LTS前，需要先注册一个华为账号并进行实名认证。如果您已有一个华为账号，可直接跳过如下操作。 进入华为云官网，单击页面右上角的“注册”。 参考 注册华为账号 并开通华为云中操作，完成注册。 注册后参考实名认证完成个人或企业账号实名认证。 为账户充值。如果您的账户有足够金额，请跳到下一个任务。如果您的账户还没有足够金额，请先为账户充值。 关于 云日志服务LTS 的价格，请参见LTS价格详情。 关于充值，请参见账户充值。 为用户添加云日志服务LTS的操作权限。 您需要有LTS的管理员权限“LTS FullAccess”，具体操作请参考授权 IAM 用户使用LTS。 本操作以Linux系统的E CS 为例，准备好需要采集日志的ECS主机，详细请参考购买弹性云服务器。如果您已有可用的ECS主机，可重复使用，不需要再次创建。

步骤一：创建日志组和日志流 日志组和日志流是云日志服务进行日志管理的基本单位，在使用云日志服务时，您首先需要创建一个日志组，然后在日志组中创建日志流。 登录云日志服务控制台。 在“日志管理”页面，单击“创建日志组”，参考表1填写日志组相关信息。 图2 创建日志组 表1 参数说明 参数 说明 示例 日志组名称 日志组名称只支持输入英文、数字、中文、中划线、下划线及小数点，且不能以小数点、下划线开头或以小数点结尾。长度为1-64个字符。 日志采集后，将发送到对应的日志组中的日志流，如果日志较多，需要分门别类，建议您给日志组做好命名，方便后续快速查找日志。 lts-group-ECS 企业项目 选择业务需要的企业项目，默认为default。 default 日志存储时间(天) 日志组的存储时间，即日志上报到LTS后日志存储的时间。 日志数据默认存储30天，可以在1~365天之间设置。 云日志服务LTS根据配置的日志存储时间定时清理日志内容，例如日志存储时间为30天，上报到LTS的日志只保存30天，30天后开始删除日志内容。 30 标签 按照业务需求对不同的日志组添加对应的标签。本示例可不设置。 - 备注 自定义填写备注信息，字符长度0-1024个字符。本示例可不填写。 - 单击“确定”，日志组创建成功，即可在日志组列表下方生成一条日志组信息。 单击目标日志组名称对应的。 单击“创建日志流”，在“创建日志流”页面中，参考表2填写日志流相关信息。 图3 创建日志流 表2 参数说明 参数 说明 示例 日志组名称 默认显示目标日志组名称。 - 日志流名称 日志流名称只支持输入英文、数字、中文、中划线、下划线及小数点，且不能以小数点、下划线开头或以小数点结尾。长度为1-64个字符。 日志采集后，以日志流为单位，将多条日志数据发往云日志服务。如果日志较多，需要分门别类，建议您创建多个日志流，并给日志流做好命名，方便后续快速查找日志。 lts-topic-ECS 企业项目 选择业务需要的企业项目，默认为default。 default 日志存储 开启日志存储：日志将会被存入搜索引擎，能使用日志全量功能。 需要开启日志存储，才能开启日志存储时间。 开启 日志存储时间(天) 日志流的存储时间，即日志上报到LTS后日志存储的时间。 日志数据默认存储30天，可以在1~365天之间设置。 打开日志流的“日志存储时间”开关：日志的存储时间使用日志流设置的日志存储时间。 云日志服务LTS根据配置的日志存储时间定时清理日志内容，例如日志存储时间为30天，上报到LTS的日志只保存30天，30天后开始删除日志内容。 30 标签 按照业务需求对不同的日志组添加对应的标签。本示例可不添加。 - 匿名写入 匿名写入默认关闭，本示例默认关闭。 匿名写入适用于安卓/IOS/小程序/浏览器端上报日志。 关闭 备注 自定义填写备注信息，字符长度0-1024个字符。本示例可不填写。 - 单击“确定”。 日志流创建成功，即可在目标日志组下方生成一条日志流信息。

同步状态说明 本节主要介绍同步完成后，各种状态的含义。方便您对异常状态进行排查定位。 以应用账号App_acc01、应用机构App_org01、原同步事件E0、新同步事件E1为例。 待处理 英文为“PENDING”，当应用账号App_acc01在生成新的同步事件E1时，如果该应用账号App_acc01之前的同步事件E0未执行完成，则生成的新事件E1的状态为待处理。 排队 英文为“QUEUING”，当应用账号App_acc01的同步事件E1可以执行时，会将该事件发送至Kafka执行消息队列，发送成功后，修改ES中存储的同步事件E1的状态为排队，表示该事件已成功加入执行队列。 发送 英文为“RUNNING”，当同步事件E1可执行时，在开始执行前将存储在ES中的同步事件E1的状态修改为RUNNING，修改成功后，调用下游应用系统的接口开始同步，表示该事件已成功处于运行状态。 成功 英文为“SUC CES S”，同步事件执行成功后，会将该事件的状态SUCCESS写入ES，表示该事件已执行成功。 失败 英文为“FAILURE”，同步事件执行失败后，会将该事件的状态FAILURE写入ES，表示该事件已执行失败。 忽视 英文为“IGNORED”，该状态涉及2种场景： 当应用账号 App_acc01 在执行一个更新的同步事件时，如果发现后续有新的更新事件，则直接忽视该事件，并执行新的更新事件。 执行全量同步时，会忽视之前所有的同步事件。 略过 英文为“SKIPPED”，该状态暂时未使用。 挂起 英文为“WAITING”，当应用账号App_acc01对应用机构App_org01存在依赖关系时，应用机构App_org01未同步成功，则该机构下应用账号App_acc01的所有同步事件状态为挂起。

前提条件 已基于Java Chassis开发好了微服务应用。 Java Chass框架下的微服务应用开发，请参考https://servicecomb.apache.org/references/java-chassis/zh_CN/。 版本要求：请参见ServiceComb引擎微服务开发框架版本要求。 本文假设您的项目使用了maven管理打包，您熟悉maven的依赖管理机制，能够正确的修改“pom.xml”文件中的dependency management和dependency。 Java Chassis支持和不同的技术进行组合使用，配置文件的名称和实际使用的技术有关。如果您采用Spring方式使用Java Chassis，配置文件的名称一般为“microservice.yaml”，如果您采用Spring Boot方式使用Java Chassis， 配置文件名称一般为“application.yaml”。本文统一使用“microservice.yaml”表示配置文件，请结合实际项目进行区分。

Spring Cloud Huawei历史版本及版本修复问题 spring-cloud-huawei版本 主要修复问题 1.11.6-2023.0.x snakeyaml、jackson、guava版本安全漏洞。 路由开关关闭后，微服务应用启动失败，找不到nacos/servicecomb适配实现类。 当前服务配置中心配置未变化，发布配置刷新事件，请求过程中可能出现池化配置找不到。 1.11.6-2022.0.x 1.11.6-2021.0.x 1.11.4-2022.0.x RBAC安全认证未开启，框架依然会监听认证过期事件。 gateway/webflux路由无法获取请求头设置信息。 微服务API安全认证开启后，规则未设置，所有请求不通过。 服务端熔断规则不生效。 1.11.4-2021.0.x 1.11.3-2022.0.x 指定服务名情况下，实例隔离策略不生效。 1.11.3-2021.0.x 1.11.2-2022.0.x API安全认证未设置黑、白名单策略时，空指针异常。 服务端、客户端同时设置相同请求头，key不生效。 1.11.2-2021.0.x 1.11.0-2022.0.x trace上下文配置基于动态配置不生效。 1.11.0-2021.0.x 1.10.13-2021.0.x 同时调用多个服务情况下，降级不生效。 1.10.11-2021.0.x 实例隔离治理不生效。 1.10.9-2021.0.x 指定服务名设置重试策略不生效。 服务降级错误返回“null”字符串，修改为返回null。 1.10.8-2021.0.x 负载均衡规则不生效。 1.10.8-2020.0.x 1.10.7-2021.0.x 服务注册发现开关关闭，启动失败。 1.10.7-2020.0.x 1.10.6-2021.0.x 监控信息中缺少环境信息。 1.10.6-2020.0.x 1.10.5-2021.0.x 重试次数太多导致请求长时间无响应。 1.10.5-2020.0.x 1.10.4-2021.0.x identifierRateLimiting限流上下文获取失败。 1.10.4-2020.0.x 1.10.3-2021.0.x 治理配置第一次变化时不生效。 1.10.3-2020.0.x 1.10.2-2021.0.x 调整默认配置刷新时间为15s。 实例隔离过滤器空指针异常。 1.10.2-2020.0.x 1.10.1-2021.0.x 非客户端请求上下文空指针异常。 路由客户端编译请求头失败。 灰度版本策略生效。 ClientRequest非RequestData类型下转化异常。 1.10.1-2020.0.x 1.10.0-2021.0.x 服务删除、重启无法刷新ribbon缓存，导致请求到不可用服务，报no host to route。 动态配置下灰度发布配置规则，修改配置规则不生效。 启动类application在业务包外层启动失败。 网关最大重试次数不生效。 1.10.0-2020.0.x 1.9.1-2020.0.x 某些场景下，启动类ags属性加载不正确。 网关配置最大重试次数无限重试。 灰度动态配置不生效。 1.9.0-2020.0.x instance.healthCheck.mode值为pull，自定义配置healthCheckInterval健康检查时间不生效。 1.8.0-2020.0.x 非long polling模式下大量配置查询任务，触发查询任务无间隔时间。 cse操作页面对服务实例下线后，服务实例调用依然正常。 management.server.port和server.port两个端口不一致，启动报错。 gateway查询到不同环境的实例。 1.7.0-2020.0.x webmvc与路由结合导致gateway启动失败。 gateway无法实现基于服务发现的路由定义功能。 gateway无法跨应用服务发现。 1.6.1-2020.0.x 说明： 存在重大问题，不建议使用： 非常频繁地查询配置中心。 查询到错误的配置。 1.9.4-Hoxton 当前服务配置中心配置未变化，发布配置刷新事件，请求过程中可能出现池化配置找不到。 1.9.3-Hoxton 服务删除、重启无法刷新ribbon缓存，导致请求到不可用服务，路由不到可用服务。 1.9.2-Hoxton 服务端删除实例后重新注册实例，客户端选择错误的服务端实例。 1.9.1-Hoxton 某些场景下，启动类ags属性加载不正确。 网关配置最大重试次数无限重试。 灰度动态配置不生效。 1.9.0-Hoxton instance.healthCheck.mode值为pull，自定义配置healthCheckInterval健康检查时间不生效。 1.8.0-Hoxton 非long polling模式下大量配置查询任务，触发查询任务无间隔时间。 cse操作页面对服务实例下线后，服务实例调用依然正常。 management.server.port和server.port两个端口不一致启动报错。 gateway查询到不同环境的实例。 1.7.0-Hoxton webmvc与路由结合导致gateway启动失败。 gateway无法实现基于服务发现的路由定义功能。 gateway无法跨应用服务发现。 1.6.0-Hoxton SDK针对401、403错误码，对引擎做出重试请求。 gateway默认路由规则不生效。 无法跨应用调用。 1.5.9-Hoxton RBAC鉴权功能不生效。 灰度路由时无法获取到微服务最新版本。 某些JDK版本不支持swagger循环依赖。 1.5.8-Hoxton gateway服务发现中选择错误实例。 实例状态为空时调用异常。 1.5.6-Hoxton AK/SK配置后不生效，鉴权失败。 server.env配置不生效，全部为空。 配置中心治理配置项删除后仍然能使用。 滑动窗口熔断配置属性不生效。 注册中心开启watch模式后，空指针异常。 读取环境变量PAAS_CSE_SC_ENDPOINT时只能读取第一个地址，无法读取第二个。 1.5.0-Hoxton 错误的治理规则导致空指针异常。 AK/SK未配置时启动，报空指针异常。 服务启动后首次并发请求，限流策略不生效。 governance治理选择错误服务导致请求异常。 当环境配置为production时，契约不变，服务重启会失败。 1.6.4-Greenwich 当前服务配置中心配置未变化，发布配置刷新事件，请求过程中可能出现池化配置找不到。 1.6.3-Greenwich 服务删除、重启无法刷新ribbon缓存，导致请求到不可用服务，路由不到可用服务。 1.6.1-Greenwich gateway跨应用服务发现异常。 1.6.0-Greenwich SDK针对401、403错误码，对引擎做出重试请求。 gateway路由规则不生效。 灰度路由时无法获取到微服务最新版本。 AK/SK配置不生效。 server.env配置不生效，全部为空。 Servicecenter在watch模式下启动失败。 读取环境变量PAAS_CSE_SC_ENDPOINT时只能读取第一个地址，无法读取第二个。 不支持跨应用调用。 1.5.0-Greenwich 错误的治理规则导致空指针异常。 AK/SK未配置时空指针异常。 服务启动后首次并发请求，限流策略不生效。 governance服务转发错误。 环境是production时，重复注册契约导致启动失败。 v1.3.3-Greenwich 注册中心监听不生效。 1.6.1-Finchley 服务启动后第一次并发，调用随机失败。 微服务跨应用调用失败。 1.6.0-Finchley SDK针对401、403错误码，对引擎做出重试请求。 gateway路由规则不生效。 灰度路由时无法获取到微服务最新版本。 AK/SK配置不生效。 server.env配置不生效，全部为空。 Servicecenter在watch模式下启动失败。 读取环境变量PAAS_CSE_SC_ENDPOINT时只能读取第一个地址，无法读取第二个。 不支持跨应用调用。 1.5.1-Finchley 配置中心治理配置删除后仍然可以使用。 v1.3.9 说明： 存在重大问题，不推荐使用。 governance存在服务转发严重错误。 v1.3.8 说明： 存在重大问题，不推荐使用。 governance存在服务转发严重错误。 v1.3.4 注册线程池无法正确关闭并导致泄漏。 actuator开启后微服务注册失败。 某些场景心跳次数过多。 v1.3.3 websocket在wss协议请求失败。 注册中心watch不生效。 v1.3.2 当环境设置为production时契约注册失败。 注册中心url未设置时随机选择地址失败。 注册中心配置 域名 时watch异常。 v1.2.0 从ServiceStage读取默认的ak/sk配置时，初始化大量对象，导致内存泄漏。 v1.1.0 heartbeat信息日志过多。 微服务间SSL调用不生效。 url包含空格时请求异常。 v1.0.0 某些场景无法自动服务发现。 v0.0.3 服务发现down状态实例。 配置的路径过长。 微服务连接本地CSE引擎失败。

开发流程说明 开发微服务应用 如果您已经完成了微服务应用的开发，可以跳过本流程，进入准备环境。 进行微服务应用开发，首先需要进行技术选型。技术选型是一个复杂的问题，技术决策者需要考虑使用的技术是否容易被团队成员掌握，技术能否满足项目对于功能、性能、可靠性方面的要求，还需要考虑商业服务等多方面的因素。本文档不探讨技术选型，假设技术团队已经选择了适合自己的开发框架。大部分技术团队都会选择开源框架来构建业务。 开发微服务应用的具体内容，请参考开发微服务应用。 使用Spring Cloud，通常会使用下面的技术进行本地微服务开发： 使用Java Chassis，通常会使用下面的技术进行本地微服务开发： 准备环境 创建云上环境，以支持ServiceComb引擎接入调试、云上应用部署和使用ServiceComb引擎功能。一般情况下，会创建一个测试环境和一个生产环境。通过ServiceStage，能够非常方便的管理云上环境，详细内容请参考准备环境。 对接微服务应用 用于微服务应用对接ServiceComb引擎，涉及到对已经开发好的应用的配置文件、构建脚本的修改。修改完成后，需要对应用重新编译、打包，通过ServiceStage将应用包部署到ServiceComb引擎，详细内容请参考对接微服务应用。 部署微服务应用 开发完成的微服务应用，通过ServiceStage部署到ServiceComb引擎，详细内容请参考部署微服务应用。 使用ServiceComb引擎功能 对于持续发展的应用系统，都会持续完善和迭代，每个迭代可能需要对微服务应用进行更新升级，需要使用更多的ServiceComb引擎功能。持续迭代的功能演进，会重复上面的应用开发、编译、打包和部署环节。详细内容请参考使用ServiceComb引擎功能。

开发能力要求 本文档的主要目的就是说明这些开源微服务开发框架如何接入和使用ServiceComb引擎的功能，假设您已经熟悉和掌握如下开发能力： 使用Java语言进行微服务开发。假设您已经基于一种ServiceStage支持的微服务开发框架开发了应用系统，并期望将应用系统托管在ServiceComb引擎上运行。本文档提供微服务应用接入ServiceComb引擎的相关技术支持。开源微服务开发框架如何使用不是本文档的范围，您可以通过开源社区获取相关微服务开发框架的入门材料和开发指南。 理解注册中心、配置中心在微服务应用中的作用，并在项目中搭建和使用注册中心。不同的微服务开发框架默认支持的开源注册中心会有差异，理解注册中心的作用，可以更加容易的更换注册中心。 熟悉应用部署，请参考创建并部署组件。

什么是Mesher Mesher是Service Mesh的一个具体的实现，是一个轻量的代理服务以Sidecar的方式与微服务一起运行。 Service Mesh是由William Morgan定义： Service Mesh是一个基础设施层，用于处理服务间通信。云原生应用有着复杂的服务拓扑，Service Mesh保证请求可以在这些拓扑中可靠地传输。在实际应用当中，Service Mesh通常是由一系列轻量级的网络代理组成的，它们与应用程序部署在一起，但应用程序不需要知道它们的存在。 随着云原生应用的崛起，Service Mesh逐渐成为一个独立的基础设施层。在云原生模型里，一个应用可以由数百个服务组成，每个服务可能有数千个实例，而每个实例可能会持续地发生变化。服务间通信不仅异常复杂，而且也是运行时行为的基础。管理好服务间通信对于保证端到端的性能和可靠性来说是非常重要的。 Service Mesh实际上就是处于TCP/IP之上的一个抽象层，假设底层的L3/L4网络能够点对点地传输字节（同时，也假设网络环境是不可靠的，所以Service Mesh必须具备处理网络故障的能力）。 从某种程度上说，Service Mesh有点类似TCP/IP。TCP对网络端点间传输字节的机制进行了抽象，而Service Mesh则是对服务节点间请求的路由机制进行了抽象。Service Mesh不关心消息体是什么，也不关心它们是如何编码的。应用程序的目标是“将某些东西从A传送到B”，而Service Mesh所要做的就是实现这个目标，并处理传送过程中可能出现的任何故障。 与TCP不同的是，Service Mesh有着更高的目标：为应用运行时提供统一的、应用层面的可见性和可控性。Service Mesh将服务间通信从底层的基础设施中分离出来，让它成为整个生态系统的一等公民——它因此可以被监控、托管和控制。

方案架构 区块链 供应链物流与现有参与方IT信息系统的结合，可实现： 统一化账本。提供不可篡改、可追溯的一致性货物流转记录，满足审计要求。 为参与方IT系统提供通用接口接入区块链服务，相关信息输入不可篡改，各自为自己的信用买单，逐步建立信任体系。 通过司机APP附加地址围栏信息自动输入实现实时展示“由谁，在什么时间，在什么地方”处理货物。 履行智能合约。实现自动签收和结算，自动运算输出绩效结果，公平公正。 图1 方案架构

行业现状及痛点 纸质单据 很多环节仍然大量沿用手工操作和纸质单据，导致运单流转时间长、成本高、对账慢、易丢失和污损，文件成本占物流成本的1/5。 效率低 供应链各协作方信息系统独立，没有统一的标准和信息系统，协作方之间协作和作用难度大。 周期长 电子信息易被篡改，只能采用纸质单据作为结算的唯一凭据，上下游核算周期长，承运商的平均应收款账期较长。 融资难 承运商大多为中小企业，缺乏信用记录，信用积累和数据，融资难，融资成本高。

方案优势 减少错误 分布式共享账本极大改善供应链的可追溯和透明度，可有效减少或消除欺诈和错误。 提高效率 通过电子POD，减少纸上作业造成的延误，通过智能合约自动结算，提高效率。 降低成本 通过快速结算，自动接单，有效跟进，可以有效降低各方的物流成本。 透明审计 不可篡改的分布式账本，不可抵赖的签名，能快速有效的发现供应链物流中存在的问题。 提高信任 货物从生产、运输到最终接收，通过区块链技术可以实现全程跟进，规则透明，自动结算，可以极大提高消费者和第三方的信任。