我遇到了与这里相同的问题并尝试解决它，但我不知道如何正确格式化数据存储，以便 cygnus 不会抛出持久性错误。
我的猎户座描述是这样的：
(curl localhost:1026/v1/subscribeContext -s -S --header 'Content-Type: application/json'
--header 'Accept: application/json' -d @- | python -mjson.tool) <<EOF
{
"entities": [
{
"type": "Event",
"isPattern": "false",
"id": "es-leon-0"
},
{
"type": "Event",
"isPattern": "false",
"id": "es-leon-1"
}
],
"attributes": [
"IdEvent", "IdUser", "Title"
],
"reference": "http://localhost:5050/notify",
"duration": "P1M",
"notifyConditions": [
{
"type": "ONCHANGE",
"condValues": [ ]
}
],
"throttling": "PT5S"
}
EOF
我的天鹅座配置：
ygnusagent.sources = http-source
cygnusagent.sinks = ckan-sink
cygnusagent.channels = ckan-channel
cygnusagent.sources.http-source.channels = ckan-channel
cygnusagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
cygnusagent.sources.http-source.port = 5050
cygnusagent.sources.http-source.handler = com.telefonica.iot.cygnus.handlers.OrionRestHandler
cygnusagent.sources.http-source.handler.notification_target = /notify
cygnusagent.sources.http-source.handler.default_service = Papel
cygnusagent.sources.http-source.handler.default_service_path = Test
cygnusagent.sources.http-source.handler.events_ttl = 5
cygnusagent.sources.http-source.interceptors = ts gi
cygnusagent.sources.http-source.interceptors.ts.type = timestamp
cygnusagent.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.GroupingInterceptor$Builder
cygnusagent.sources.http-source.interceptors.gi.gropuing_rules_conf_file = /Applications/apache-flume-1.4.0-bin/conf/grouping_rules.conf
cygnusagent.channels.ckan-channel.type = memory
cygnusagent.channels.ckan-channel.capacity = 1000
cygnusagent.channels.ckan-channel.transactionCapacity = 100
# ============================================
# OrionCKANSink configuration
# channel name from where to read notification events
cygnusagent.sinks.ckan-sink.channel = ckan-channel
# sink class, must not be changed
cygnusagent.sinks.ckan-sink.type = com.telefonica.iot.cygnus.sinks.OrionCKANSink
# true if the grouping feature is enabled for this sink, false otherwise
cygnusagent.sinks.ckan-sink.enable_grouping = false
# true if lower case is wanted to forced in all the element names, false otherwise
cygnusagent.sinks.hdfs-sink.enable_lowercase = false
# the CKAN API key to use
cygnusagent.sinks.ckan-sink.api_key = xxxxx
# the FQDN/IP address for the CKAN API endpoint
cygnusagent.sinks.ckan-sink.ckan_host = ckan-demo.ckan.io
# the port for the CKAN API endpoint
cygnusagent.sinks.ckan-sink.ckan_port = 80
# Orion URL used to compose the resource URL with the convenience operation URL to query it
cygnusagent.sinks.ckan-sink.orion_url = http://localhost:1026
# how the attributes are stored, either per row either per column (row, column)
cygnusagent.sinks.ckan-sink.attr_persistence = column
# enable SSL for secure Http transportation; 'true' or 'false'
cygnusagent.sinks.ckan-sink.ssl = false
# number of notifications to be included within a processing batch
cygnusagent.sinks.ckan-sink.batch_size = 100
# timeout for batch accumulation
cygnusagent.sinks.ckan-sink.batch_timeout = 60
# number of retries upon persistence error
cygnusagent.sinks.ckan-sink.batch_ttl = 10
天鹅座正在接收它，但随后显示以下错误：
time=2016-04-21T07:44:57.504CDT | lvl=INFO | trans=1461242686-614-0000000001 | srv=Papel | subsrv=Test | function=getEvents | comp=Cygnus | msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[231] : Starting transaction (1461242686-614-0000000001)
time=2016-04-21T07:44:57.528CDT | lvl=INFO | trans=1461242686-614-0000000001 | srv=Papel | subsrv=Test | function=getEvents | comp=Cygnus | msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[258] : Received data ({ "subscriptionId" : "571897360e94f9fa53829885", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "type" : "Event", "isPattern" : "false", "id" : "es-leon-0", "attributes" : [ { "name" : "IdEvent", "type" : "text", "value" : "1084" }, { "name" : "IdUser", "type" : "text", "value" : "18" }, { "name" : "Title", "type" : "text", "value" : "Papes" } ] }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
time=2016-04-21T07:44:57.528CDT | lvl=INFO | trans=1461242686-614-0000000001 | srv=Papel | subsrv=Test | function=getEvents | comp=Cygnus | msg=com.telefonica.iot.cygnus.handlers.OrionRestHandler[280] : Event put in the channel, id=2024732986
time=2016-04-21T07:45:50.771CDT | lvl=INFO | trans=1461242686-614-0000000001 | srv=Papel | subsrv=Test | function=persistAggregation | comp=Cygnus | msg=com.telefonica.iot.cygnus.sinks.OrionCKANSink[417] : [ckan-sink] Persisting data at OrionCKANSink (orgName=papel, pkgName=papel_test, resName=es-leon-0_event, data={"recvTime": "2016-04-21T12:44:57.497Z","fiwareServicePath": "Test","entityId": "es-leon-0","entityType": "Event","Title": "Papes"},{"recvTime": "2016-04-21T12:44:57.528Z","fiwareServicePath": "Test","entityId": "es-leon-0","entityType": "Event","IdEvent": "1084","IdUser": "18","Title": "Papes"})
time=2016-04-21T07:45:51.875CDT | lvl=ERROR | trans=1461242686-614-0000000001 | srv=Papel | subsrv=Test | function=processNewBatches | comp=Cygnus | msg=com.telefonica.iot.cygnus.sinks.OrionSink[426] : Runtime error (Cannot persist the data (orgName=papel, pkgName=papel_test, resName=es-leon-0_event))
如此处所述，我创建了相应的数据存储：http://ckan-demo.ckan.io/dataset/papel-test/resource/8d7cb489-878e-465e-8c8c-60ea537411e0但是不知道如何格式化它或csv是否正确格式。
谢谢
*注意：我在行模式下尝试过并且一切正常，但这不是我想要的。
**注意：我还在预览器软件中发现一个错误，将我的列"标题"的标题更改为页面标题"CKAN 演示"。
编辑：
我已经完成了文档中所说的操作：
列：为所有通知的上下文属性更新一行。这种行将包含每个实体属性的两个字段（一个用于值，称为<attrName>，另一个用于元数据，称为<attrName>_md），以及四个附加字段：
- recvTime：人类可编辑格式的 UTC 时间戳 （ISO 8601）。
- fiwareServicePath：通知的路径或默认路径。 实体
- ID：通知的实体标识符。 实体类型
- ：通知的实体类型。
但仍然有相同的错误：
time=2016-04-25T05:17:48.790CDT | lvl=ERROR | trans=1461579403-571-0000000000 | srv=Papel | subsrv=Test | function=processNewBatches | comp=Cygnus | msg=com.telefonica.iot.cygnus.sinks.OrionSink[426] : Runtime error (Cannot persist the data (orgName=papel, pkgName=papel_test, resName=es-leon-0_event))
首先，在创建资源和关联的数据存储之前，您需要一个 CKAN 组织和包/数据集，以便持久保存数据。
创建一个组织，假设在 demo.ckan.org 中;组织名称是 frb ，因为我们的实体将位于该 FIWARE 服务中：
$ curl -X POST "http://demo.ckan.org/api/3/action/organization_create" -d '{"name":"frb"}' -H "Authorization: xxxxxxxx"
在上述组织内创建一个包/数据集;包名称是frb_test，因为我们的实体将位于 FIWARE 服务frb和 FIWARE 服务路径中 test：
$ curl -X POST "http://demo.ckan.org/api/3/action/package_create" -d '{"name":"frb_test","owner_org":"frb"}' -H "Authorization: xxxxxxxx"
在上述包/数据集中创建资源（包 ID 在对上述包创建请求的响应中给出）;资源的名称room1_room，因为实体 ID 将被room1，其类型room：
$ curl -X POST "http://demo.ckan.org/api/3/action/resource_create" -d '{"name":"room1_room","url":"none","format":"","package_id":"d35fca28-732f-4096-8376-944563f175ba"}' -H "Authorization: xxxxxxxx"
最后，回答您的问题，创建一个与上述资源关联的数据存储，适合在列模式下接收 Cgynus 数据（资源 ID 在对上述资源创建请求的响应中给出）：
$ curl -X POST "http://demo.ckan.org/api/3/action/datastore_create" -d '{"fields":[{"id":"recvTime","type":"text"}, {"id":"fiwareServicePath","type":"text"}, {"id":"entityId","type":"text"}, {"id":"entityType","type":"text"}, {"id":"temperature","type":"float"}, {"id":"temperature_md","type":"json"}],"resource_id":"48c120df-5bcd-48c7-81fa-8ecf4e4ef9d7","force":"true"}' -H "Authorization: xxxxxxxx"
现在，Cygnus 能够在 frb 服务中持久化 ID room1 类型为 room 的实体的数据，test服务路径：
time=2016-04-26T15:54:45.753CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=getEvents | comp=cygnusagent | msg=com.telefonica.iot.cygnus.handlers.NGSIRestHandler[240] : Starting internal transaction (b465ffb8-710f-4cd3-9573-dc3799f774f9)
time=2016-04-26T15:54:45.754CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=getEvents | comp=cygnusagent | msg=com.telefonica.iot.cygnus.handlers.NGSIRestHandler[256] : Received data ({ "subscriptionId" : "51c0ac9ed714fb3b37d7d5a8", "originator" : "localhost", "contextResponses" : [ { "contextElement" : { "attributes" : [ { "name" : "temperature", "type" : "centigrade", "value" : "26.5" } ], "type" : "room", "isPattern" : "false", "id" : "room1" }, "statusCode" : { "code" : "200", "reasonPhrase" : "OK" } } ]})
time=2016-04-26T15:55:07.843CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=processNewBatches | comp=cygnusagent | msg=com.telefonica.iot.cygnus.sinks.NGSISink[342] : Batch accumulation time reached, the batch will be processed as it is
time=2016-04-26T15:55:07.844CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=processNewBatches | comp=cygnusagent | msg=com.telefonica.iot.cygnus.sinks.NGSISink[396] : Batch completed, persisting it
time=2016-04-26T15:55:07.846CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=persistAggregation | comp=cygnusagent | msg=com.telefonica.iot.cygnus.sinks.NGSICKANSink[419] : [ckan-sink] Persisting data at OrionCKANSink (orgName=frb, pkgName=frb_test, resName=room1_room, data={"recvTime": "2016-04-26T13:54:45.756Z","fiwareServicePath": "/test","entityId": "room1","entityType": "room","temperature": "26.5"})
time=2016-04-26T15:55:08.948CEST | lvl=INFO | corr=b465ffb8-710f-4cd3-9573-dc3799f774f9 | trans=b465ffb8-710f-4cd3-9573-dc3799f774f9 | svc=frb | subsvc=/test | function=processNewBatches | comp=cygnusagent | msg=com.telefonica.iot.cygnus.sinks.NGSISink[400] : Finishing internal transaction (b465ffb8-710f-4cd3-9573-dc3799f774f9)
插入也可以通过 CKAN API 进行检查：
$ curl -X POST "http://demo.ckan.org/api/3/action/datastore_search" -d '{"resource_id":"48c120df-5bcd-48c7-81fa-8ecf4e4ef9d7"}' -H "Authorization: xxxxxxxx"