持久数据宇宙不起作用
Persisting data cosmos not working
我试图将在 Orion 中收到的信息保存在 Cosmos 的 public 实例中，但我收到了一个 http 500 错误。这是 contextBroker 日志：
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[154]: Starting transaction to localhost:5060/notify
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[364]: Notification Successfully Sent to localhost:5060/notify
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[375]: Transaction ended
function=connectionTreat | comp=Orion | msg=rest.cpp[767]: Starting transaction from 192.168.201.95:49160/v1/updateContext
function=processContextElement | comp=Orion | msg=MongoCommonUpdate.cpp[1647]: Database Operation Successful ({ _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES", _id.servicePath: /^\/$|^\/\/.*/ })
function=addTriggeredSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[922]: Database Operation Successful ({ $or: [ { entities.id: "613cb7464a7950ba", $or: [ { entities.type: "UMANICK_TECHNOLOGIES" }, { entities.type: { $exists: false } } ], entities.isPattern: "false", conditions.type: "ONCHANGE", conditions.value: "X_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] } }, { entities.isPattern: "true", conditions.type: "ONCHANGE", conditions.value: "X_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] }, $where: function(){for (var i=0; i < this.entities.length; i++) {if (this.enti... } ] })
function=addTriggeredSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[922]: Database Operation Successful ({ $or: [ { entities.id: "613cb7464a7950ba", $or: [ { entities.type: "UMANICK_TECHNOLOGIES" }, { entities.type: { $exists: false } } ], entities.isPattern: "false", conditions.type: "ONCHANGE", conditions.value: "Z_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] } }, { entities.isPattern: "true", conditions.type: "ONCHANGE", conditions.value: "Z_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] }, $where: function(){for (var i=0; i < this.entities.length; i++) {if (this.enti... } ] })
function=processContextElement | comp=Orion | msg=MongoCommonUpdate.cpp[1834]: Database Operation Successful (update { _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES", _id.servicePath: /\// })
function=entitiesQuery | comp=Orion | msg=MongoGlobal.cpp[1090]: Database Operation Successful ({ query: { $or: [ { _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES" } ], _id.servicePath: { $in: [ null, /^/$/ ] }, attrs.name: { $in: [ "X_VALUE", "Y_VALUE", "Z_VALUE" ] } }, orderby: { creDate: 1 } })
function=processSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[1046]: Database Operation Successful (update: { $set: { lastNotification: 1435574837 }, $inc: { count: 1 } }, query: { _id: ObjectId('55912126a85d63ed38229b6d') })
function=requestCompleted | comp=Orion | msg=rest.cpp[382]: Transaction ended
订阅 Orion 正常工作，Cygnus 代理收到通知，但是当尝试在 Cosmos 中持久化数据时出现 http 500 错误。
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.OrionHDFSSink.persist:212) - Persisting data. File: 613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt, Data: 2015-06-29T12:50:18.763|1435575018|613cb7464a7950ba|UMANICK_TECHNOLOGIES|X_VALUE|float|0.30645782)
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.exists:158) - HttpFS operation: GET http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=getfilestatus&user.name=ilko.garcia HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.exists:161) - HttpFS response: HTTP/1.1 200 OK
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:123) - HttpFS operation: POST http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=append&user.name=ilko.garcia HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:126) - HttpFS response: HTTP/1.1 307 Temporary Redirect
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:138) - HttpFS operation: POST http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=append&user.name=ilko.garcia&data=true HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:141) - HttpFS response: HTTP/1.1 500 Internal Server Error
我的Cygnus代理配置文件如下：
# APACHE_FLUME_HOME/conf/cygnus.conf
orionagent.sources = http-source
orionagent.sinks = hdfs-sink
orionagent.channels = notifications
# Flume source, must not be changed
orionagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
# channel name where to write the notification events
orionagent.sources.http-source.channels = notifications
# listening port the Flume source will use for receiving incoming notifications
orionagent.sources.http-source.port = 5060
# Flume handler that will parse the notifications, must not be changed
orionagent.sources.http-source.handler = es.tid.fiware.orionconnectors.cosmosinjector.OrionRestHandler
# regular expression for the orion version the notifications will have in their headers
orionagent.sources.http-source.handler.orion_version = 0\.19\.*
# URL target
orionagent.sources.http-source.handler.notification_target = /notify
# channel name from where to read notification events
orionagent.sinks.hdfs-sink.channel = notifications
# Flume sink that will process and persist in HDFS the notification events, must not be changed
orionagent.sinks.hdfs-sink.type = es.tid.fiware.orionconnectors.cosmosinjector.OrionHDFSSink
# IP address of the Cosmos deployment where the notification events will be persisted
orionagent.sinks.hdfs-sink.cosmos_host = 130.206.80.46
# port of the Cosmos service listening for persistence operations; 14000 for httpfs, 50070 for webhdfs and free choice for inifinty
orionagent.sinks.hdfs-sink.cosmos_port = 14000
# username allowed to write in HDFS (/user/myusername)
orionagent.sinks.hdfs-sink.cosmos_username = ilko.garcia
orionagent.sinks.hdfs-sink.cosmos_default_password = *************
# dataset where to persist the data (/user/myusername/mydataset)
orionagent.sinks.hdfs-sink.cosmos_dataset = mydataset
# HDFS backend type (webhdfs, httpfs or infinity)
orionagent.sinks.hdfs-sink.hdfs_api = httpfs
# how the attributes are stored, either per row either per column (row, column)
orionagent.sinks.hdfs-sink.attr_persistence = column
# channel name
orionagent.channels.notifications.type = memory
# capacity of the channel
orionagent.channels.notifications.capacity = 1000
# amount of bytes that can be sent per transaction
orionagent.channels.notifications.transactionCapacity = 100
我的 Orion 实例在 spain2 区域。
提前致谢。
============================================= =========================
编辑 1
我安装了新版本的 Cygnus (0.8.1)，但没有捕捉到创建的引导代理设置。
这里可以看到日志flume和配置文件Cygnus。
flume.log
msg=org.mortbay.log.Slf4jLog[67] : Logging to org.slf4j.impl.Log4jLoggerAdapter(org.mortbay.log) via org.mortbay.log.Slf4jLog
msg=com.telefonica.iot.cygnus.nodes.CygnusApplication[238] : Starting Cygnus application
msg=org.mortbay.log.Slf4jLog[67] : jetty-6.1.26
msg=org.apache.flume.node.PollingPropertiesFileConfigurationProvider[61] : Configuration provider starting
msg=org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable[133] : Reloading configuration file:/usr/cygnus/conf/agent_umanick.conf
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[930] : Added sinks: hdfs-sink Agent: cygnusagent
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[336] : Agent configuration for 'cygnusagent' does not contain any valid channels. Marking it as invalid.
msg=org.apache.flume.conf.FlumeConfiguration[127] : Agent configuration invalid for agent 'cygnusagent'. It will be removed.
msg=org.apache.flume.conf.FlumeConfiguration[140] : Post-validation flume configuration contains configuration for agents: []
msg=org.apache.flume.node.AbstractConfigurationProvider[138] : No configuration found for this host:cygnusagent
msg=org.apache.flume.node.Application[138] : Starting new configuration:{ sourceRunners:{} sinkRunners:{} channels:{} }
msg=org.mortbay.log.Slf4jLog[67] : Started SocketConnector@0.0.0.0:8081
msg=org.apache.flume.node.Application[101] : Shutting down configuration: { sourceRunners:{} sinkRunners:{} channels:{} }
msg=org.apache.flume.node.Application[138] : Starting new configuration:{ sourceRunners:{} sinkRunners:{} channels:{} }
agent_umanick.conf
#=============================================
# To be put in APACHE_FLUME_HOME/conf/cygnus.conf
#
# General configuration template explaining how to setup a sink of each of the available types (HDFS, CKAN, MySQL).
#=============================================
# The next tree fields set the sources, sinks and channels used by Cygnus. You could use different names than the
# ones suggested below, but in that case make sure you keep coherence in properties names along the configuration file.
# Regarding sinks, you can use multiple types at the same time; the only requirement is to provide a channel for each
# one of them (this example shows how to configure 3 sink types at the same time). Even, you can define more than one
# sink of the same type and sharing the channel in order to improve the performance (this is like having
# multi-threading).
cygnusagent.sources = http-source
cygnusagent.sinks = hdfs-sink
cygnusagent.channels = hdfs-channel
#=============================================
# source configuration
# channel name where to write the notification events
cygnusagent.sources.http-source.channels = hdfs-channel
# source class, must not be changed
cygnusagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
# listening port the Flume source will use for receiving incoming notifications
cygnusagent.sources.http-source.port = 5050
# Flume handler that will parse the notifications, must not be changed
cygnusagent.sources.http-source.handler = com.telefonica.iot.cygnus.handlers.OrionRestHandler
# URL target
cygnusagent.sources.http-source.handler.notification_target = /notify
# Number of channel re-injection retries before a Flume event is definitely discarded (-1 means infinite retries)
cygnusagent.sources.http-source.handler.events_ttl = 10
# ============================================
# OrionHDFSSink configuration
# channel name from where to read notification events
cygnusagent.sinks.hdfs-sink.channel = hdfs-channel
# sink class, must not be changed
cygnusagent.sinks.hdfs-sink.type = com.telefonica.iot.cygnus.sinks.OrionHDFSSink
# Comma-separated list of FQDN/IP address regarding the HDFS Namenode endpoints
# If you are using Kerberos authentication, then the usage of FQDNs instead of IP addresses is mandatory
cygnusagent.sinks.hdfs-sink.hdfs_host = 130.206.80.46
# port of the HDFS service listening for persistence operations; 14000 for httpfs, 50070 for webhdfs
cygnusagent.sinks.hdfs-sink.hdfs_port = 14000
# username allowed to write in HDFS
cygnusagent.sinks.hdfs-sink.hdfs_username = ilko.garcia
# password for the username
cygnusagent.sinks.hdfs-sink.hdfs_password = *************
# how the attributes are stored, either per row either per column (row, column)
cygnusagent.sinks.hdfs-sink.attr_persistence = column
cygnus_instance_umanick.conf
# The OS user that will be running Cygnus. Note this must be `root` if you want to run cygnus in a privileged port (<1024), either the admin port or the port in which Cygnus receives Orion notifications
CYGNUS_USER=cygnus
# Which is the config folder
CONFIG_FOLDER=/usr/cygnus/conf
# Which is the config file
CONFIG_FILE=/usr/cygnus/conf/agent_umanick.conf
# Name of the agent. The name of the agent is not trivial, since it is the base for the Flume parameters naming conventions, e.g. it appears in <AGENT_NAME>.sources.http-source.channels=...
AGENT_NAME=orion
# Name of the logfile located at /var/log/cygnus. It is important to put the extension '.log' in order to the log rotation works properly
LOGFILE_NAME=cygnus.log
# Administration port. Must be unique per instance
ADMIN_PORT=8081
# Polling interval (seconds) for the configuration reloading
POLLING_INTERVAL=30
============================================= =========================
编辑 2
我添加了缺少的属性并且代理正常启动。但是数据仍然没有持久化在 cosmos 中。代理配置相同，缺少添加的属性
flume.log
msg=org.apache.flume.node.Application[145] : Starting Channel hdfs-channel
msg=org.apache.flume.instrumentation.MonitoredCounterGroup[94] : Component type: CHANNEL, name: hdfs-channel started
msg=org.apache.flume.node.Application[173] : Starting Sink hdfs-sink
msg=org.apache.flume.node.Application[184] : Starting Source http-source
msg=org.mortbay.log.Slf4jLog[67] : jetty-6.1.26
msg=org.mortbay.log.Slf4jLog[67] : Started SocketConnector@0.0.0.0:5050
msg=org.apache.flume.instrumentation.MonitoredCounterGroup[94] : Component type: SOURCE, name: http-source started
msg=com.telefonica.iot.cygnus.backends.http.HttpClientFactory[79] : Setting max total connections (500)
msg=com.telefonica.iot.cygnus.backends.http.HttpClientFactory[80] : Settubg default max connections per route (100)
msg=com.telefonica.iot.cygnus.sinks.OrionHDFSSink[257] : [hdfs-sink] Startup completed
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[150] : Starting transaction (1435685260-766-0000000000)
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[236] : Received data ({ "subscriptionId" : "5592bbe2a85d63e738d33bac", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "type" : "UMANICK_TECHNOLOGIES_S_L", "isPattern" : "false", "id" : "613cb7464a7950ba", "attributes" : [ { "name" : "X_VALUE", "type" : "float", "value" : "0.15322891" }, { "name" : "Y_VALUE", "type" : "float", "value" : "-0.076614454" }, { "name" : "Z_VALUE", "type" : "float", "value" : "9.844957" } ] }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[258] : Event put in the channel (id=220543398, ttl=10)
msg=com.telefonica.iot.cygnus.sinks.OrionSink[128] : Event got from the channel (id=220543398, headers={content-type=application/json, transactionId=1435685260-766-0000000000, fiware-service=def_serv, fiware-servicepath=def_serv_path, ttl=10}, bodyLength=722)
msg=com.telefonica.iot.cygnus.sinks.OrionSink[184] : null
msg=com.telefonica.iot.cygnus.sinks.OrionSink[193] : Finishing transaction (1435685260-766-0000000000)
============================================= =========================
编辑 3
flume.log
Starting transaction (1435739504-848-0000000005)
Received data ({ "subscriptionId" : "5592bbe2a85d63e738d33bac", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "type" : "UMANICK_TECHNOLOGIES_S_L", "isPattern" : "false", "id" : "613cb7464a7950ba", "attributes" : [ { "name" : "X_VALUE", "type" : "float", "value" : "0.22984336" }, { "name" : "Y_VALUE", "type" : "float", "value" : "-0.11492168" }, { "name" : "Z_VALUE", "type" : "float", "value" : "9.921572" } ] }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
Event put in the channel (id=1449000407, ttl=10)
Event got from the channel (id=1449000407, headers={timestamp=1435739601153, content-type=application/json, transactionId=1435739504-848-0000000005, fiware-service=def_serv, fiware-servicepath=def_servpath, ttl=10, destination=613cb7464a7950ba_umanick_technologies_s_l}, bodyLength=721)
Persisting data at OrionHDFSSink. HDFS file (def_serv/def_servpath/613cb7464a7950ba_umanick_technologies_s_l/613cb7464a7950ba_umanick_technologies_s_l.txt), Data ({"recvTime":"2015-07-01T08:33:21.153Z","X_VALUE":"0.22984336", "X_VALUE_md":[],"Y_VALUE":"-0.11492168", "Y_VALUE_md":[],"Z_VALUE":"9.921572", "Z_VALUE_md":[]})
Finishing transaction (1435739504-848-0000000005)
看来您使用的是相当旧的 Cygnus 版本。 es.tid.fiware.orionconnectors.cosmosinjector...等包在前段时间被替换了
请参考 https://github.com/telefonicaid/fiware-cygnus (it matches 0.8.1) or configure the FIWARE repo in order to get the RPM for 0.7.1 as described at https://github.com/telefonicaid/fiware-cygnus#section4.1
主分支中的最新版本
编辑 1
请注意这个日志：
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[336] : Agent configuration for 'cygnusagent' does not contain any valid channels. Marking it as invalid.
如果您查看 agent_umanick.conf 文件，您会发现它缺少如下内容：
cygnusagent.sinks.hdfs-channel.type = memory
cygnusagent.sinks.hdfs-channel.capacity = 10000
cygnusagent.sinks.hdfs-channel.transactionCapacity = 100
在https://github.com/telefonicaid/fiware-cygnus/blob/master/README.md#section5.2
查看
编辑 2
同样，如果您查看 https://github.com/telefonicaid/fiware-cygnus/blob/master/README.md#section5.2，您会发现源和接收器中缺少一些重要的配置参数：
# Default service (service semantic depends on the persistence sink)
cygnusagent.sources.http-source.handler.default_service = def_serv
# Default service path (service path semantic depends on the persistence sink)
cygnusagent.sources.http-source.handler.default_service_path = def_servpath
# Source interceptors, do not change
cygnusagent.sources.http-source.interceptors = ts gi
# TimestampInterceptor, do not change
cygnusagent.sources.http-source.interceptors.ts.type = timestamp
# GroupingInterceptor, do not change
cygnusagent.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.GroupingInterceptor$Builder
# Grouping rules for the GroupingIntercetor, put the right absolute path to the file if necessary
# See the doc/design/interceptors document for more details
cygnusagent.sources.http-source.interceptors.gi.grouping_rules_conf_file = /usr/cygnus/conf/grouping_rules.conf
# Hive FQDN/IP address of the Hive server
cygnusagent.sinks.hdfs-sink.hive_host = cosmos.lab.fi-ware.org
# Hive port for Hive external table provisioning
cygnusagent.sinks.hdfs-sink.hive_port = 10000
# Kerberos-based authentication enabling
cygnusagent.sinks.hdfs-sink.krb5_auth = false
其中，非常重要的是/usr/cygnus/conf/grouping_rules.conf存在，哪怕是空的。该软件提供了一个名为grouping_rules.conf.template的模板；请 rename/copy 它以获得适当的文件。
编辑 3
为了读取持久数据而要执行的命令：
hadoop fs -cat /user/ilko.garcia/def_serv/def_servpath/613cb7464a7950ba_umanick_technologies_s_l/613cb7464a7950ba_umanick_technologies_s_l.txt
我试图将在 Orion 中收到的信息保存在 Cosmos 的 public 实例中，但我收到了一个 http 500 错误。这是 contextBroker 日志：
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[154]: Starting transaction to localhost:5060/notify
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[364]: Notification Successfully Sent to localhost:5060/notify
function=sendHttpSocket | comp=Orion | msg=clientSocketHttp.cpp[375]: Transaction ended
function=connectionTreat | comp=Orion | msg=rest.cpp[767]: Starting transaction from 192.168.201.95:49160/v1/updateContext
function=processContextElement | comp=Orion | msg=MongoCommonUpdate.cpp[1647]: Database Operation Successful ({ _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES", _id.servicePath: /^\/$|^\/\/.*/ })
function=addTriggeredSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[922]: Database Operation Successful ({ $or: [ { entities.id: "613cb7464a7950ba", $or: [ { entities.type: "UMANICK_TECHNOLOGIES" }, { entities.type: { $exists: false } } ], entities.isPattern: "false", conditions.type: "ONCHANGE", conditions.value: "X_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] } }, { entities.isPattern: "true", conditions.type: "ONCHANGE", conditions.value: "X_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] }, $where: function(){for (var i=0; i < this.entities.length; i++) {if (this.enti... } ] })
function=addTriggeredSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[922]: Database Operation Successful ({ $or: [ { entities.id: "613cb7464a7950ba", $or: [ { entities.type: "UMANICK_TECHNOLOGIES" }, { entities.type: { $exists: false } } ], entities.isPattern: "false", conditions.type: "ONCHANGE", conditions.value: "Z_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] } }, { entities.isPattern: "true", conditions.type: "ONCHANGE", conditions.value: "Z_VALUE", expiration: { $gt: 1435574837 }, servicePath: { $in: [ /^$|^/#$|^/$/, null ] }, $where: function(){for (var i=0; i < this.entities.length; i++) {if (this.enti... } ] })
function=processContextElement | comp=Orion | msg=MongoCommonUpdate.cpp[1834]: Database Operation Successful (update { _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES", _id.servicePath: /\// })
function=entitiesQuery | comp=Orion | msg=MongoGlobal.cpp[1090]: Database Operation Successful ({ query: { $or: [ { _id.id: "613cb7464a7950ba", _id.type: "UMANICK_TECHNOLOGIES" } ], _id.servicePath: { $in: [ null, /^/$/ ] }, attrs.name: { $in: [ "X_VALUE", "Y_VALUE", "Z_VALUE" ] } }, orderby: { creDate: 1 } })
function=processSubscriptions | comp=Orion | msg=MongoCommonUpdate.cpp[1046]: Database Operation Successful (update: { $set: { lastNotification: 1435574837 }, $inc: { count: 1 } }, query: { _id: ObjectId('55912126a85d63ed38229b6d') })
function=requestCompleted | comp=Orion | msg=rest.cpp[382]: Transaction ended
订阅 Orion 正常工作，Cygnus 代理收到通知，但是当尝试在 Cosmos 中持久化数据时出现 http 500 错误。
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.OrionHDFSSink.persist:212) - Persisting data. File: 613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt, Data: 2015-06-29T12:50:18.763|1435575018|613cb7464a7950ba|UMANICK_TECHNOLOGIES|X_VALUE|float|0.30645782)
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.exists:158) - HttpFS operation: GET http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=getfilestatus&user.name=ilko.garcia HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.exists:161) - HttpFS response: HTTP/1.1 200 OK
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:123) - HttpFS operation: POST http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=append&user.name=ilko.garcia HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:126) - HttpFS response: HTTP/1.1 307 Temporary Redirect
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:138) - HttpFS operation: POST http://130.206.80.46:14000/webhdfs/v1/user/ilko.garcia/mydataset/613cb7464a7950ba-UMANICK_TECHNOLOGIES-X_VALUE-float.txt?op=append&user.name=ilko.garcia&data=true HTTP/1.1
[SinkRunner-PollingRunner-DefaultSinkProcessor] (es.tid.fiware.orionconnectors.cosmosinjector.hdfs.HttpFSBackend.append:141) - HttpFS response: HTTP/1.1 500 Internal Server Error
我的Cygnus代理配置文件如下：
# APACHE_FLUME_HOME/conf/cygnus.conf
orionagent.sources = http-source
orionagent.sinks = hdfs-sink
orionagent.channels = notifications
# Flume source, must not be changed
orionagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
# channel name where to write the notification events
orionagent.sources.http-source.channels = notifications
# listening port the Flume source will use for receiving incoming notifications
orionagent.sources.http-source.port = 5060
# Flume handler that will parse the notifications, must not be changed
orionagent.sources.http-source.handler = es.tid.fiware.orionconnectors.cosmosinjector.OrionRestHandler
# regular expression for the orion version the notifications will have in their headers
orionagent.sources.http-source.handler.orion_version = 0\.19\.*
# URL target
orionagent.sources.http-source.handler.notification_target = /notify
# channel name from where to read notification events
orionagent.sinks.hdfs-sink.channel = notifications
# Flume sink that will process and persist in HDFS the notification events, must not be changed
orionagent.sinks.hdfs-sink.type = es.tid.fiware.orionconnectors.cosmosinjector.OrionHDFSSink
# IP address of the Cosmos deployment where the notification events will be persisted
orionagent.sinks.hdfs-sink.cosmos_host = 130.206.80.46
# port of the Cosmos service listening for persistence operations; 14000 for httpfs, 50070 for webhdfs and free choice for inifinty
orionagent.sinks.hdfs-sink.cosmos_port = 14000
# username allowed to write in HDFS (/user/myusername)
orionagent.sinks.hdfs-sink.cosmos_username = ilko.garcia
orionagent.sinks.hdfs-sink.cosmos_default_password = *************
# dataset where to persist the data (/user/myusername/mydataset)
orionagent.sinks.hdfs-sink.cosmos_dataset = mydataset
# HDFS backend type (webhdfs, httpfs or infinity)
orionagent.sinks.hdfs-sink.hdfs_api = httpfs
# how the attributes are stored, either per row either per column (row, column)
orionagent.sinks.hdfs-sink.attr_persistence = column
# channel name
orionagent.channels.notifications.type = memory
# capacity of the channel
orionagent.channels.notifications.capacity = 1000
# amount of bytes that can be sent per transaction
orionagent.channels.notifications.transactionCapacity = 100
我的 Orion 实例在 spain2 区域。
提前致谢。
============================================= ========================= 编辑 1
我安装了新版本的 Cygnus (0.8.1)，但没有捕捉到创建的引导代理设置。
这里可以看到日志flume和配置文件Cygnus。
flume.log
msg=org.mortbay.log.Slf4jLog[67] : Logging to org.slf4j.impl.Log4jLoggerAdapter(org.mortbay.log) via org.mortbay.log.Slf4jLog
msg=com.telefonica.iot.cygnus.nodes.CygnusApplication[238] : Starting Cygnus application
msg=org.mortbay.log.Slf4jLog[67] : jetty-6.1.26
msg=org.apache.flume.node.PollingPropertiesFileConfigurationProvider[61] : Configuration provider starting
msg=org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable[133] : Reloading configuration file:/usr/cygnus/conf/agent_umanick.conf
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[930] : Added sinks: hdfs-sink Agent: cygnusagent
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[1016] : Processing:hdfs-sink
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[336] : Agent configuration for 'cygnusagent' does not contain any valid channels. Marking it as invalid.
msg=org.apache.flume.conf.FlumeConfiguration[127] : Agent configuration invalid for agent 'cygnusagent'. It will be removed.
msg=org.apache.flume.conf.FlumeConfiguration[140] : Post-validation flume configuration contains configuration for agents: []
msg=org.apache.flume.node.AbstractConfigurationProvider[138] : No configuration found for this host:cygnusagent
msg=org.apache.flume.node.Application[138] : Starting new configuration:{ sourceRunners:{} sinkRunners:{} channels:{} }
msg=org.mortbay.log.Slf4jLog[67] : Started SocketConnector@0.0.0.0:8081
msg=org.apache.flume.node.Application[101] : Shutting down configuration: { sourceRunners:{} sinkRunners:{} channels:{} }
msg=org.apache.flume.node.Application[138] : Starting new configuration:{ sourceRunners:{} sinkRunners:{} channels:{} }
agent_umanick.conf
#=============================================
# To be put in APACHE_FLUME_HOME/conf/cygnus.conf
#
# General configuration template explaining how to setup a sink of each of the available types (HDFS, CKAN, MySQL).
#=============================================
# The next tree fields set the sources, sinks and channels used by Cygnus. You could use different names than the
# ones suggested below, but in that case make sure you keep coherence in properties names along the configuration file.
# Regarding sinks, you can use multiple types at the same time; the only requirement is to provide a channel for each
# one of them (this example shows how to configure 3 sink types at the same time). Even, you can define more than one
# sink of the same type and sharing the channel in order to improve the performance (this is like having
# multi-threading).
cygnusagent.sources = http-source
cygnusagent.sinks = hdfs-sink
cygnusagent.channels = hdfs-channel
#=============================================
# source configuration
# channel name where to write the notification events
cygnusagent.sources.http-source.channels = hdfs-channel
# source class, must not be changed
cygnusagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
# listening port the Flume source will use for receiving incoming notifications
cygnusagent.sources.http-source.port = 5050
# Flume handler that will parse the notifications, must not be changed
cygnusagent.sources.http-source.handler = com.telefonica.iot.cygnus.handlers.OrionRestHandler
# URL target
cygnusagent.sources.http-source.handler.notification_target = /notify
# Number of channel re-injection retries before a Flume event is definitely discarded (-1 means infinite retries)
cygnusagent.sources.http-source.handler.events_ttl = 10
# ============================================
# OrionHDFSSink configuration
# channel name from where to read notification events
cygnusagent.sinks.hdfs-sink.channel = hdfs-channel
# sink class, must not be changed
cygnusagent.sinks.hdfs-sink.type = com.telefonica.iot.cygnus.sinks.OrionHDFSSink
# Comma-separated list of FQDN/IP address regarding the HDFS Namenode endpoints
# If you are using Kerberos authentication, then the usage of FQDNs instead of IP addresses is mandatory
cygnusagent.sinks.hdfs-sink.hdfs_host = 130.206.80.46
# port of the HDFS service listening for persistence operations; 14000 for httpfs, 50070 for webhdfs
cygnusagent.sinks.hdfs-sink.hdfs_port = 14000
# username allowed to write in HDFS
cygnusagent.sinks.hdfs-sink.hdfs_username = ilko.garcia
# password for the username
cygnusagent.sinks.hdfs-sink.hdfs_password = *************
# how the attributes are stored, either per row either per column (row, column)
cygnusagent.sinks.hdfs-sink.attr_persistence = column
cygnus_instance_umanick.conf
# The OS user that will be running Cygnus. Note this must be `root` if you want to run cygnus in a privileged port (<1024), either the admin port or the port in which Cygnus receives Orion notifications
CYGNUS_USER=cygnus
# Which is the config folder
CONFIG_FOLDER=/usr/cygnus/conf
# Which is the config file
CONFIG_FILE=/usr/cygnus/conf/agent_umanick.conf
# Name of the agent. The name of the agent is not trivial, since it is the base for the Flume parameters naming conventions, e.g. it appears in <AGENT_NAME>.sources.http-source.channels=...
AGENT_NAME=orion
# Name of the logfile located at /var/log/cygnus. It is important to put the extension '.log' in order to the log rotation works properly
LOGFILE_NAME=cygnus.log
# Administration port. Must be unique per instance
ADMIN_PORT=8081
# Polling interval (seconds) for the configuration reloading
POLLING_INTERVAL=30
============================================= ========================= 编辑 2
我添加了缺少的属性并且代理正常启动。但是数据仍然没有持久化在 cosmos 中。代理配置相同，缺少添加的属性
flume.log
msg=org.apache.flume.node.Application[145] : Starting Channel hdfs-channel
msg=org.apache.flume.instrumentation.MonitoredCounterGroup[94] : Component type: CHANNEL, name: hdfs-channel started
msg=org.apache.flume.node.Application[173] : Starting Sink hdfs-sink
msg=org.apache.flume.node.Application[184] : Starting Source http-source
msg=org.mortbay.log.Slf4jLog[67] : jetty-6.1.26
msg=org.mortbay.log.Slf4jLog[67] : Started SocketConnector@0.0.0.0:5050
msg=org.apache.flume.instrumentation.MonitoredCounterGroup[94] : Component type: SOURCE, name: http-source started
msg=com.telefonica.iot.cygnus.backends.http.HttpClientFactory[79] : Setting max total connections (500)
msg=com.telefonica.iot.cygnus.backends.http.HttpClientFactory[80] : Settubg default max connections per route (100)
msg=com.telefonica.iot.cygnus.sinks.OrionHDFSSink[257] : [hdfs-sink] Startup completed
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[150] : Starting transaction (1435685260-766-0000000000)
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[236] : Received data ({ "subscriptionId" : "5592bbe2a85d63e738d33bac", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "type" : "UMANICK_TECHNOLOGIES_S_L", "isPattern" : "false", "id" : "613cb7464a7950ba", "attributes" : [ { "name" : "X_VALUE", "type" : "float", "value" : "0.15322891" }, { "name" : "Y_VALUE", "type" : "float", "value" : "-0.076614454" }, { "name" : "Z_VALUE", "type" : "float", "value" : "9.844957" } ] }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[258] : Event put in the channel (id=220543398, ttl=10)
msg=com.telefonica.iot.cygnus.sinks.OrionSink[128] : Event got from the channel (id=220543398, headers={content-type=application/json, transactionId=1435685260-766-0000000000, fiware-service=def_serv, fiware-servicepath=def_serv_path, ttl=10}, bodyLength=722)
msg=com.telefonica.iot.cygnus.sinks.OrionSink[184] : null
msg=com.telefonica.iot.cygnus.sinks.OrionSink[193] : Finishing transaction (1435685260-766-0000000000)
============================================= ========================= 编辑 3
flume.log
Starting transaction (1435739504-848-0000000005)
Received data ({ "subscriptionId" : "5592bbe2a85d63e738d33bac", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "type" : "UMANICK_TECHNOLOGIES_S_L", "isPattern" : "false", "id" : "613cb7464a7950ba", "attributes" : [ { "name" : "X_VALUE", "type" : "float", "value" : "0.22984336" }, { "name" : "Y_VALUE", "type" : "float", "value" : "-0.11492168" }, { "name" : "Z_VALUE", "type" : "float", "value" : "9.921572" } ] }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
Event put in the channel (id=1449000407, ttl=10)
Event got from the channel (id=1449000407, headers={timestamp=1435739601153, content-type=application/json, transactionId=1435739504-848-0000000005, fiware-service=def_serv, fiware-servicepath=def_servpath, ttl=10, destination=613cb7464a7950ba_umanick_technologies_s_l}, bodyLength=721)
Persisting data at OrionHDFSSink. HDFS file (def_serv/def_servpath/613cb7464a7950ba_umanick_technologies_s_l/613cb7464a7950ba_umanick_technologies_s_l.txt), Data ({"recvTime":"2015-07-01T08:33:21.153Z","X_VALUE":"0.22984336", "X_VALUE_md":[],"Y_VALUE":"-0.11492168", "Y_VALUE_md":[],"Z_VALUE":"9.921572", "Z_VALUE_md":[]})
Finishing transaction (1435739504-848-0000000005)
看来您使用的是相当旧的 Cygnus 版本。 es.tid.fiware.orionconnectors.cosmosinjector...等包在前段时间被替换了
请参考 https://github.com/telefonicaid/fiware-cygnus (it matches 0.8.1) or configure the FIWARE repo in order to get the RPM for 0.7.1 as described at https://github.com/telefonicaid/fiware-cygnus#section4.1
主分支中的最新版本编辑 1
请注意这个日志：
msg=org.apache.flume.conf.FlumeConfiguration$AgentConfiguration[336] : Agent configuration for 'cygnusagent' does not contain any valid channels. Marking it as invalid.
如果您查看 agent_umanick.conf 文件，您会发现它缺少如下内容：
cygnusagent.sinks.hdfs-channel.type = memory
cygnusagent.sinks.hdfs-channel.capacity = 10000
cygnusagent.sinks.hdfs-channel.transactionCapacity = 100
在https://github.com/telefonicaid/fiware-cygnus/blob/master/README.md#section5.2
查看编辑 2
同样，如果您查看 https://github.com/telefonicaid/fiware-cygnus/blob/master/README.md#section5.2，您会发现源和接收器中缺少一些重要的配置参数：
# Default service (service semantic depends on the persistence sink)
cygnusagent.sources.http-source.handler.default_service = def_serv
# Default service path (service path semantic depends on the persistence sink)
cygnusagent.sources.http-source.handler.default_service_path = def_servpath
# Source interceptors, do not change
cygnusagent.sources.http-source.interceptors = ts gi
# TimestampInterceptor, do not change
cygnusagent.sources.http-source.interceptors.ts.type = timestamp
# GroupingInterceptor, do not change
cygnusagent.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.GroupingInterceptor$Builder
# Grouping rules for the GroupingIntercetor, put the right absolute path to the file if necessary
# See the doc/design/interceptors document for more details
cygnusagent.sources.http-source.interceptors.gi.grouping_rules_conf_file = /usr/cygnus/conf/grouping_rules.conf
# Hive FQDN/IP address of the Hive server
cygnusagent.sinks.hdfs-sink.hive_host = cosmos.lab.fi-ware.org
# Hive port for Hive external table provisioning
cygnusagent.sinks.hdfs-sink.hive_port = 10000
# Kerberos-based authentication enabling
cygnusagent.sinks.hdfs-sink.krb5_auth = false
其中，非常重要的是/usr/cygnus/conf/grouping_rules.conf存在，哪怕是空的。该软件提供了一个名为grouping_rules.conf.template的模板；请 rename/copy 它以获得适当的文件。
编辑 3
为了读取持久数据而要执行的命令：
hadoop fs -cat /user/ilko.garcia/def_serv/def_servpath/613cb7464a7950ba_umanick_technologies_s_l/613cb7464a7950ba_umanick_technologies_s_l.txt