mysql - 我的 Cygnus 代理没有创建 MySQL 数据库

Question

下午好!

遵循 http://fiware-cygnus.readthedocs.io 的文档我创建了我的两个 conf 如下:

cygnus_instance_1.conf(编辑:更新于 2017/11/23)

CYGNUS_USER=cygnus
CONFIG_FOLDER=/usr/cygnus/conf
CONFIG_FILE=/usr/cygnus/conf/agent_1.conf
AGENT_NAME=cygnus-ngsi
LOGFILE_NAME=cygnus.log
ADMIN_PORT=8081
POLLING_INTERVAL=30

agent_1.conf(编辑:更新于 2017/12/1)

cygnus-ngsi.sources = http-source
cygnus-ngsi.sinks = mysql-sink
cygnus-ngsi.channels = mysql-channel

cygnus-ngsi.sources.http-source.channels = mysql-channel
cygnus-ngsi.sources.http-source.type = org.apache.flume.source.http.HTTPSource
cygnus-ngsi.sources.http-source.port = 5050
cygnus-ngsi.sources.http-source.handler = com.telefonica.iot.cygnus.handlers.NGSIRestHandler
cygnus-ngsi.sources.http-source.handler.notification_target = /notify
cygnus-ngsi.sources.http-source.handler.default_service =
cygnus-ngsi.sources.http-source.handler.default_service_path = /
cygnus-ngsi.sources.http-source.handler.events_ttl = 2
cygnus-ngsi.sources.http-source.interceptors = ts
cygnus-ngsi.sources.http-source.interceptors.ts.type = timestamp

cygnus-ngsi.channels.mysql-channel.type = memory
cygnus-ngsi.channels.mysql-channel.capacity = 1000
cygnus-ngsi.channels.mysql-channel.transactionCapacity = 100


cygnus-ngsi.sinks.mysql-sink.mysql_port = 3306
cygnus-ngsi.sinks.mysql-sink.mysql_host = localhost
cygnus-ngsi.sinks.mysql-sink.enable_grouping = true
cygnus-ngsi.sinks.mysql-sink.mysql_password = *********
cygnus-ngsi.sinks.mysql-sink.type = com.telefonica.iot.cygnus.sinks.NGSIMySQLSink
cygnus-ngsi.sinks.mysql-sink.batch_timeout = 10
cygnus-ngsi.sinks.mysql-sink.mysql_username = username
cygnus-ngsi.sinks.mysql-sink.batch_size = 1
cygnus-ngsi.sinks.mysql-sink.channel = mysql-channel
cygnus-ngsi.sinks.mysql-sink.new_parameter = new_value
cygnus-ngsi.sinks.mysql-sink.data_model = dm-by-attribute
cygnus-ngsi.sinks.mysql-sink.attr_persistence = row

我之前创建了我的订阅，以便在更改事件时，之前的实体保存在 MySQL 中。

    $ch = curl_init();

curl_setopt($ch, CURLOPT_URL, "http://$IP_PUBLIC:1026/v2/subscriptions");
curl_setopt($ch, CURLOPT_RETURNTRANSFER, TRUE);
curl_setopt($ch, CURLOPT_HEADER, FALSE);

curl_setopt($ch, CURLOPT_POST, TRUE);

curl_setopt($ch, CURLOPT_POSTFIELDS, "{
  \"description\": \"One subscription to rule them all\",
  \"subject\": {
    \"entities\": [
      {
        \"idPattern\": \".*\",
        \"type\": \"ExerciseAction\"
      }
    ],
    \"condition\": {
      \"attrs\": [
        \"startTime\"
      ]
    }
  },
  \"notification\": {
    \"http\": {
      \"url\": \"http://$IP_PUBLIC:5050/notify\"
    },
    \"attrs\": [
      \"startTime\" 
    ]
  },
  \"expires\": \"2020-04-05T14:00:00.00Z\"
}");

curl_setopt($ch, CURLOPT_HTTPHEADER, array(
  "Content-Type: application/json"
));

$response = curl_exec($ch);
curl_close($ch);

var_dump($response);

cygnus.log 更新实体后

time=2017-12-01T12:52:50.539Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:52:53.031Z | lvl=INFO | corr=8b61fc74-d696-11e7-a933-fa163ec68c31 | trans=ccb60fea-10f8-46eb-82fa-8864f9352e41 | srv=default | subsrv=/ | comp=cygnus-ngsi | op=getEvents | 

msg=com.telefonica.iot.cygnus.handlers.NGSIRestHandler[286] : [NGSIRestHandler] Starting internal transaction (ccb60fea-10f8-46eb-82fa-8864f9352e41)
time=2017-12-01T12:52:53.042Z | lvl=INFO | corr=8b61fc74-d696-11e7-a933-fa163ec68c31 | trans=ccb60fea-10f8-46eb-82fa-8864f9352e41 | srv=default | subsrv=/ | comp=cygnus-ngsi | op=getEvents | 

msg=com.telefonica.iot.cygnus.handlers.NGSIRestHandler[304] : [NGSIRestHandler] Received data ({  "subscriptionId" : "5a2150a4d3258a6d7c656d26",  "originator" : "localhost",  "contextResponses" : [    {      "contextElement" : {        

"type" : "ExerciseAction",        "isPattern" : "false",        "id" : "75",        "attributes" : [          {            "name" : "startTime",            "type" : "Text",            "value" : "30/11/2017 14:38:40"          }        ]  

    },      "statusCode" : {        "code" : "200",        "reasonPhrase" : "OK"      }    }  ]})
time=2017-12-01T12:52:55.546Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:00.548Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:05.550Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:10.551Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:15.554Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:20.558Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:25.559Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:30.561Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:35.568Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:40.569Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:45.570Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:50.571Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:53:55.574Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:54:00.575Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:54:05.576Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:54:10.580Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)
time=2017-12-01T12:54:15.588Z | lvl=ERROR | corr=cdeca432-d695-11e7-9c51-fa163ec68c31 | trans=fe9234ec-055b-483d-876d-e15d442a0efe | srv=default | subsrv=/ | comp=cygnus-ngsi | op=run | msg=org.apache.flume.SinkRunner$PollingRunner[160] 

: Unable to deliver event. Exception follows.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:476)
        at com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:368)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:748)

有人可以帮助我并告诉我为什么我的 Cygnus 和 MySQL 之间的连接不起作用吗？非常感谢!!!

Answer 1

根据您的日志，配置中存在错误。这一行:

cygnus-ngsi.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.NGSIGroupingInterceptor

必须是:

cygnus-ngsi.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.NGSIGroupingInterceptor$Builder

此外，您必须授予此变量内的文件存在:

cygnus-ngsi.sources.http-source.interceptors.gi.grouping_rules_conf_file = /Applications/apache-flume-1.4.0-bin/conf/grouping_rules.conf

无论如何，这是为了使用分组拦截，这是一项高级功能(也是一项已弃用的功能；建议使用名称映射)。因此，我的建议是删除这些行，只保留时间戳拦截:

cygnus-ngsi.sources.http-source.interceptors = ts
cygnus-ngsi.sources.http-source.interceptors.ts.type = timestamp