流处理模型

流数据服务(Streaming Service)以流处理模型作为服务来源，其中指定了运行服务所需的信息。

流数据处理流程包含：

依据上述处理流程，流数据的处理模型包含四个部分：Receiver（接收器）、Filter（过滤器）、Mapper（转换器）和Sender（发送器）。每个部分作为一个节点，可以进行连接和合并，构建成实时数据处理流 Stream。除了处理流 Stream 以外，还有一些辅助参数作为整个服务的运行条件，一并存储在启动参数类型 Startup 中。处理模型如下图：

 配置参数

流处理模型采用 JSON 格式定义，您可以参考下文介绍的参数及相应的 JSON 示例，编写一个流处理模型文件并发布为流数据服务。您也可以使用流处理模型编辑器构建模型，查看参数说明即可。

SparkParameter

用于设置 Spark Streaming 的运行参数。包括：

• checkPointDir：设置 Streaming 的CheckPoint功能的保存目录。String 类型。
• interval：int 类型。设置 Streaming 运行的间隔时间，单位为毫秒。

Stream

Stream 中包含了实时数据处理运行流的参数。

• nodeDic：存放运行的节点的词典。
• StreamNode：运行流节点基类，记录节点的基本信息如name、source和description，以及最重要的节点运行的先后顺序。采用prevNodes记录当前节点的前序节点列表，nextNodes记录当前节点后序节点列表。 prevNodes 和 nextNodes 的值是一个数组，数组元素是节点的名称，即每个节点可以有多个前序节点，也可以有多个后序节点。对于 Receiver 节点，prevNodes 为空；对于Sender 节点，nextNodes 为空。 如果某个节点有多个前序节点，那么，这些前序节点的 metadata 必须相同，否则，执行会报错。

Receiver

继承自StreamNode，作为流数据处理的入口，接收各种来源的数据，包括Socket、WebSocket、Http、文件系统等。Receiver中需要设置接收信息的元数据，即metadata。Receiver节点包括三个部分组成：自身的描述信息如name、source等；消息的元数据metadata；消息的读取格式reader。

流数据服务支持以下接收方式：

SocketReceiver：继承自Receiver，接收Socket消息的节点。需指定的参数有：

•  ipAddress——String 类型。接收的Socket服务的IP地址

•  port——int 类型，接收的Socket服务的端口号

示例：

{
   "ipAddress" : "127.0.0.1",
   "port" : 9527, 
   "name" : "socketReceiver", 
   "source" : "Socket Receiver", 
   "description" : "Receive some message from socketServer", 
   "prevNodes" : [],
   "nextNodes" : [],
   "className": "com.supermap.bdt.streaming.receiver.SocketReceiver "
}

MultiSocketReceiver：继承自Receiver，同时接收多个Socket消息的节点，接收的消息内容必须是相同的。需指定的参数有：

• servers——Array[String] 类型。需要接收的多个服务地址，每个数组对象为一个地址，地址与端口用冒号隔开。

示例：

{
  "servers": [
    "192.168.1.1:9527",
    "192.168.1.1:9528",
    "192.168.1.2:9527"
  ],
    "name": "multiSocketReceiver",
    "source": "MultiSource Socket Receiver",
    "description": "Receive message from multi socket server",
    "prevNodes": [],
    "nextNodes": [],
    "className": "com.supermap.bdt.streaming.receiver.MultiSocketReceiver"
}

  SocketServerReceiver：继承自Receiver，Socket服务端接收节点，用于作为服务端接收其他Socket客户的发送的消息。需指定的参数有：

• port：int型。启动的Socket服务端监听端口。

{
  "port": 9527,
  "name": "socketServerReceiver",
  "source": "SocketServer Receiver",
  "description": "Receive message from socket client",
  "prevNodes": [],
  "nextNodes": [],
"className": "com.supermap.bdt.streaming.receiver.SocketServerReceiver"
}

  WebSocketReceiver：继承自Receiver，接收WebSocket消息的节点。需指定的参数有：

• url：String 类型。WebSocket服务地址。

{
  "url": "ws://192.168.1.1:9527/websocket ",
  "name": "webSocketReceiver",
  "source": "WebSocket Receiver",
  "description": "Receive message from websocket server",
  "prevNodes": [],
  "nextNodes": [],
  "className": "com.supermap.bdt.streaming.receiver.WebSocketReceiver"
}

  TextFileReceiver：继承自Receiver，监控指定目录，读取新增文件的内容。需指定的参数有：

• directoryPath：监控的文件目录，如HDFS目录hdfs:///data/；Linux系统中的目录 /user/share/data；Windows系统中的目录C:/data。

{
  "directoryPath": "'hdfs:///data/'",
  "name": "textFileReceiver",
  "source": "Text File Receiver",
  "description": "Listen new file in folder",
  "prevNodes": [],
  "nextNodes": [],
  "className": "com.supermap.bdt.streaming.receiver.TextFileReceiver"
}

SingleTextFileReceiver：单文本文件接收器，继承自Receiver，根据设置读取监控文件的内容，支持读取 Json、GeoJSON 和 CSV格式的文件。需指定的参数有：

• readInterva：读取时间的间隔。

• rowsOneTime：每次读取的行数。

{
  "version": 9000,
  "sparkParameter": {
    "checkPointDir": "tmp",
    "interval": 5000
  },
  "stream": {
    "nodeDic": {
      "TextFileReceiver": {
        "filePath": "G:\\QQRev\\test.json",
        "readInterva": 1000,
        "rowsOneTime": 100,
        "reader": {
          "isJsonArray": false,
          "arrayExpression": "",
          "className": "com.supermap.bdt.streaming.formatter.JsonFormatter"
        },
        "metadata": {
          "title": "",
          "epsg": 3857,
          "fieldInfos": [
            {
              "name": "X",
              "source": "lon",
              "nType": "DOUBLE"
            },
            {
              "name": "Y",
              "source": "lat",
              "nType": "DOUBLE"
            },
            {
              "name": "mbbh",
              "source": "mbbh",
              "nType": "TEXT"
            }
          ],
          "featureType": "POINT",
          "idFieldName": "mbbh",
          "dateTimeFormat": "yyyy-MM-dd HH:mm:ss"
        },
        "name": "TextFileReceiver",
        "caption": "",
        "description": "",
        "prevNodes": [],
        "nextNodes": [
          "ConsoleSender"
        ],
        "className": "com.supermap.bdt.streaming.receiver.SingleTextFileReceiver"
      },
      "ConsoleSender": {
        "formatter": {
          "separator": ",",
          "className": "com.supermap.bdt.streaming.formatter.CSVFormatter"
        },
        "name": "ConsoleSender",
        "caption": "",
        "description": "",
        "prevNodes": [
          "TextFileReceiver"
        ],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.sender.ConsoleSender"
      }
    }
  }
}

KafkaReceiver：继承自Receiver，接收kafka消息的节点。需指定的参数有：

• servers——String 类型。服务器列表，如果有多个服务器之间用逗号分隔，端口用分号“;”分隔
• topics——Array[String] 类型。kafka topic 数组
• groupid—— String 类型。kafka 的 groupID
• offset——String类型。kafka的读取位置，默认为latest，即只读取最新消息

{  "servers": "192.168.1.1:9092, 192.168.1.2:9092, 192.168.1.3:9092 ,192.168.1.4:9092",
  "topics": [
    "topic1",
    "topic2"
  ],
  "groupid": "groupId", 
  "offset": "latest",
  "name": "kafkaReceiver",
  "source": "Kafka Receiver",
  "description": "Receive message from Kafka",
  "prevNodes": [],
  "nextNodes": [],
"className": "com.supermap.bdt.streaming.receiver.KafkaReceiver"
}

HttpReceiver： 继承自Receiver，接收 HTTP 的消息节点，目前只支持HTTP的Get方法。

• url：String类型。Http服务地址。

{
  "url": "https://api.wheretheiss.at/v1/satellites/25544",
  "name": "httpReceiver",
  "source": "HTTP Receiver",
  "description": "Get message from web",
  "prevNodes": [],
  "nextNodes": [],
  "className": "com.supermap.bdt.streaming.receiver.HttpReceiver"
}

JMSReceiver：继承自Receiver，接收JMS标准协议消息的节点，用于接收ActiveMQ、RabbitMQ等消息中间件的消息。

• url——JMS消息服务地址

• port——int类型。消息服务端口

• queueName ——String类型。 消息队列名称

• jdniName——String类型。对应消息中间件的JDNI名称，需要到中间件官网查询

• username——String类型。用户名

• password——String类型。密码

{
  "url": "192.168.1.1",
  "port": 9527,
  "queueName": "data",
  "jdniName": "org.apache.activemq.jndi.ActiveMQInitialContextFactory",
  "userName": "user",
  "password": "password",
  "name": "jmsReceiver",
  "source": "JMS Receiver",
  "description": "Receive message from JMS(Java Message Service) for ActiveMQ",
  "prevNodes": [],
  "nextNodes": [],
"className": "com.supermap.bdt.streaming.receiver.JMSReceiver"
}

metadata

metadata 写在 Receiver 参数中，是接收消息的元数据，用于描述消息的格式定义。需指定以下信息：

• title：元数据的名称，用于区分其他元数据。String类型
• featureType：FeatureType类。接收消息转换的地理要素类型，如点POINT、线LINE、面REGION等。
• epsg：int 类型。元数据地理要素的投影EPSG编码。
• fieldInfos：接收消息转换后的字段信息。需指定：

• name：String类型。字段名称，为字段的唯一标识
• source：String类型。字段在原始信息中的位置，决定了从原始信息中的什么位置去解析成为本字段的值。成为本字段的值。如果原始信息为 CSV 格式，source 值为 CSV 中的字段序号，如 "source": "4" 代表了 CSV 数据中的第 5 个字段；如果原始数据为 json 格式的，那么 source 值为 json 中键值对的键。
• nType：FieldType 类型。字段的类型，如：字符型 TEXT、双精度浮点型 DOUBLE、整型 INT 等。

reader

接收的消息的内容格式，包括CSV格式（CSVFormatter）、JSON格式（JsonFormatter）或者GeoJSON格式（GeoJsonFormatter）。

CSVFormatter：表示接收的消息的内容格式为CSV格式。需指定：

• separator：指定分隔符，默认为逗号

"reader": {
    "separator": ",",
    "className": "com.supermap.bdt.streaming.formatter.CSVFormatter"
  }

JsonFormatter：接收的消息内容格式为 JSON。示例如下：

"reader": {
    "className": "com.supermap.bdt.streaming.formatter.JsonFormatter"
  }

GeoJsonFormatter：接收的消息内容格式为 GeoJSON。示例如下：

"reader": {
    "className": "com.supermap.bdt.streaming.formatter.JsonFormatter"
  }

Filter

继承于StreamNode，用于过滤当前数据，进行数据的清洗与整理。需指定：

• filter：如果需要获取字段值进行逻辑运算，使用关键词[]，如[ID] > 10。多个运算式直接可以使用&&（与）或者 ||（或）进行连接，并使用括号()进行优先顺序调整，如[ID] > 10 && ([X] >= 10 || [Y] <= 65.32)。注意，使用关键词IN、MATCHES、EXISTS、ISNULL与其他运算式一起进行运算时，必须使用括号包括起来，如 ([ID] IN 1,3,5,7,9) && [X] > 100。

Mapper

继承于StreamNode，用于建立字段映射以及对字段进行管理，主要包括：字段映射、添加字段、删除字段、字段运算以及地理围栏。

添加字段

• insertIndex——int类型。字段添加的位置
• fieldname——String类型。添加的字段名称
• nType——SFieldType类型。添加的字段类型
• expression——String类型。添加字段值可以为已有字段的运算表达式的值：

"insertMapper": {
        "insertIndex": 1,
        "fieldName": "XX",
        "nType": "DOUBLE",
        "expression": "[X] * 2",
        "name": "insertMapper",
        "source": "Insert Field",
        "description": "Insert Field by X * 2",
        "prevNodes": [],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.map.FeatureInsertMapper"
      }

删除字段

• deleteFieldNames：删除字段名称数组

"deleteMapper": {
        "deleteFieldNames": [
    "F1",
    "F2"
  ],
        "name": "deleteMapper",
        "source": "delete Field",
        "description": "delete Field F1和F2 ,
        "prevNodes": [],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.map.FeatureDeleteMapper"
      }

字段映射

• srcToDesIndexPair：源数据中的字段（字段在流中的序号）与名称的对照关系

• srcToDesNamePair：字段名称与新名称的对照关系

"mapMaper": {
        "srcToDesNamePair": {
          "ID": "newID_Name",
          "Y": "newY_Name",
          "X": "newX_Name"
        },
        "srcToDesIndexPair": {
          "ID": 0,
          "Y": 2,
          "X": 1
        },
        "name": "mapMaper",
        "source": "Map Fields",
        "description": "Map Fields with new name and index",
        "prevNodes": [],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.map.FeatureMapMapper"
      }

字段运算

• fieldName：目标字段名称
• expression：字段的运算表达式

"calculateMapper": {
        "fieldName ": Fcal,
        "expression": "[X] * 2",
        "name": "calculateMapper",
        "source": "calculate Field",
        "description": "calculate Field by X * 2",
        "prevNodes": [],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.map.FeatureCalculateMapper "
      }

地理围栏

• connection：地理围栏对象的来源。

• type——String类型。数据源类型

• Info——Array[DsInfo]类型。数据源连接信息。

• • server：数据服务器
  • datasetNames：用Array[String]类型。用于地理围栏的数据集名称
•  fenceName——String类型。进入地理围栏对象的Name属性字段。
• fenceID——String类型。进入地理围栏对象的ID字段，也就是唯一标识该对象的字段。
• withinFieldName——String类型。新增字段的字段名称，该字段用于记录当前对象是否在地理围栏内。
• statusFieldName——String类型。新增字段的字段名称，该字段用于记录当前对象的状态是进入地理围栏还是离开地理围栏。

"GeoFenceMapper": {
        "connection": {
          "type": "udb",
          "info": [
            {
              "server": "Z: \\airport.udb",
              "datasetNames": [
                "airports_40"
              ]
            }
          ]
        },
        "fenceName": "NAME",
        "fenceID": "SmID",
        "withinFieldName": "geoWithin",
        "statusFieldName": "geoStatus",
        "name": "GeoFenceMapper",
        "source": "地理标记转换",
        "description": "",
        "prevNodes": [
          "SocketReceiver"
        ],
        "nextNodes": [
          "GeoJsonSocketSender",
          "FenceWithinFilterOut",
          "FenceWithinFilterIn"
        ],
        "className": "com.supermap.bdt.streaming.map.GeoTaggerMapper"
      }

Sender

继承于StreamNode，作为流数据处理的出口，向外发送数据。包含：

• EsAppendSender
• EsUpdateSender
• FileSender
• JMSSender
• SenderBase
• SMSSender
• SocketClientSender
• SocketServerSender
• WebSocketClientSender
• WebSocketTool

以常用的 WebSocket 协议为例，发送数据配置如下：

"webSocketSender": {
        "path": "ws://127.0.0.1/data",
        "name": "webSocketSender",
        "source": "WebSocket Sender",
        "description": "Send message to WebSocket Server",
        "prevNodes": [],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.sender.WebSocketClientSender"
      }

以输出至 Elasticsearch 为例：

      "ESSender": {
        "url": "192.168.168.33",
        "queueName": "aircondition",
        "directoryPath": "test1",
        "bitmap$0": false,
        "formatter": {
          "className": "com.supermap.bdt.streaming.formatter.GeoJsonFormatter"
        },
        "name": "ESSender",
        "source": "ES发送",
        "description": "",
        "prevNodes": [
          "TextFileReceiverJson"
        ],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.sender.EsAppendSender"
      }

参照上述示例设置，可以将数据的分析处理结果输出到 iServer DataStore 创建的时空数据库中。

编写流处理模型文件

根据上述参数说明，编写一个完整的流处理模型文件，您可以将文件另存为后缀为.streaming的文件，用于快速发布流数据服务，也可以将流处理模型文件的内容直接写入“配置信息”中进行发布。示例如下：

{
  "version": 9000,
  "sparkParameter": {
    "checkPointDir": "tmp",
    "interval": 10000
  },
  "stream": {
    "nodeDic": {
      "AQIReceiver": {
        "url": "http://www.supermapol.com/iserver/services/aqi/restjsr/aqi/pm2_5.json?bounds=-113.90625001585,-52.029966847235,113.90625001585,69.175579762077&to=910111",
        "reader": {
          "isJsonArray": true,
          "arrayExpression": "airQualityList",
          "className": "com.supermap.bdt.streaming.formatter.JsonFormatter"
        },
        "metadata": {
          "title": "",
          "epsg": 3857,
          "fieldInfos": [
            {
              "name": "X",
              "source": "location.x",
              "nType": "DOUBLE"
            },
            {
              "name": "Y",
              "source": "location.y",
              "nType": "DOUBLE"
            },
            {
              "name": "positionName",
              "source": "positionName",
              "nType": "TEXT"
            },
            {
              "name": "aqi",
              "source": "aqi",
              "nType": "DOUBLE"
            }
          ],
          "featureType": "POINT"
        },
        "name": "AQIReceiver",
        "caption": "",
        "description": "",
        "prevNodes": [],
        "nextNodes": [
          "WebSocketClientSender"
        ],
        "className": "com.supermap.bdt.streaming.receiver.HttpReceiver"
      },
      "WebSocketClientSender": {
        "path": "ws://127.0.0.1:8800/iserver/services/dataflow/dataflow/broadcast",
        "formatter": {
          "separator": ",",
          "className": "com.supermap.bdt.streaming.formatter.GeoJsonFormatter"
        },
        "name": "WebSocketClientSender",
        "caption": "",
        "description": "",
        "prevNodes": [
          "AQIReceiver"
        ],
        "nextNodes": [],
        "className": "com.supermap.bdt.streaming.sender.WebSocketClientSender"
      }
    }
  }
}