Как хранить данные в MySql с помощью cygnus?

Question

Я прочитал всю документацию о том, как работает cygnus, я специально проверил this one. Я также закончил читать this учебник, но я уверен, что я не настроил что-то правильно.

в cygnus_instance_1.conf я создал:

CYGNUS_USER=root 
CONFIG_FOLDER=/usr/cygnus/conf 
CONFIG_FILE=/usr/cygnus/conf/agent_1.conf 
AGENT_NAME=cygnusagent 
LOGFILE_NAME=cygnus.log 
ADMIN_PORT=8081 

и agent_1.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 mysql-sink ckan-sink 
cygnusagent.channels = hdfs-channel mysql-channel ckan-channel 

#============================================= 
# source configuration 
# channel name where to write the notification events 
cygnusagent.sources.http-source.channels = hdfs-channel mysql-channel ckan-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 = es.tid.fiware.fiwareconnectors.cygnus.handlers.OrionRestHandler 
# URL target 
cygnusagent.sources.http-source.handler.notification_target = /notify 
# 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 
# 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 
# Source interceptors, do not change 
cygnusagent.sources.http-source.interceptors = ts de 
# Interceptor type, do not change 
cygnusagent.sources.http-source.interceptors.ts.type = timestamp 
# Destination extractor interceptor, do not change 
cygnusagent.sources.http-source.interceptors.de.type = es.tid.fiware.fiwareconnectors.cygnus.interceptors.DestinationExtractor$Builder 
# Matching table for the destination extractor interceptor, put the right absolute path to the file if necessary 
# See the doc/design/interceptors document for more details 
cygnusagent.sources.http-source.interceptors.de.matching_table = /usr/cygnus/conf/matching_table.conf 

# ============================================ 
# 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 = es.tid.fiware.fiwareconnectors.cygnus.sinks.OrionHDFSSink 
# Comma-separated list of FQDN/IP address regarding the Cosmos Namenode endpoints 
# If you are using Kerberos authentication, then the usage of FQDNs instead of IP addresses is mandatory 
cygnusagent.sinks.hdfs-sink.cosmos_host = x1.y1.z1.w1,x2.y2.z2.w2 
# port of the Cosmos service listening for persistence operations; 14000 for httpfs, 50070 for webhdfs and free choice for inifinty 
cygnusagent.sinks.hdfs-sink.cosmos_port = 14000 
# default username allowed to write in HDFS 
cygnusagent.sinks.hdfs-sink.cosmos_default_username = cosmos_username 
# default password for the default username 
cygnusagent.sinks.hdfs-sink.cosmos_default_password = xxxxxxxxxxxxx 
# HDFS backend type (webhdfs, httpfs or infinity) 
cygnusagent.sinks.hdfs-sink.hdfs_api = httpfs 
# how the attributes are stored, either per row either per column (row, column) 
cygnusagent.sinks.hdfs-sink.attr_persistence = column 
# Hive FQDN/IP address of the Hive server 
cygnusagent.sinks.hdfs-sink.hive_host = x.y.z.w 
# 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 
# Kerberos username 
cygnusagent.sinks.hdfs-sink.krb5_auth.krb5_user = krb5_username 
# Kerberos password 
cygnusagent.sinks.hdfs-sink.krb5_auth.krb5_password = xxxxxxxxxxxxx 
# Kerberos login file 
cygnusagent.sinks.hdfs-sink.krb5_auth.krb5_login_conf_file = /usr/cygnus/conf/krb5_login.conf 
# Kerberos configuration file 
cygnusagent.sinks.hdfs-sink.krb5_auth.krb5_conf_file = /usr/cygnus/conf/krb5.conf 

# ============================================ 
# 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 = es.tid.fiware.fiwareconnectors.cygnus.sinks.OrionCKANSink 
# the CKAN API key to use 
cygnusagent.sinks.ckan-sink.api_key = ckanapikey 
# the FQDN/IP address for the CKAN API endpoint 
cygnusagent.sinks.ckan-sink.ckan_host = x.y.z.w 
# 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 = row 
# enable SSL for secure Http transportation; 'true' or 'false' 
cygnusagent.sinks.ckan-sink.ssl = false 

# ============================================ 
# OrionMySQLSink configuration 
# channel name from where to read notification events 
cygnusagent.sinks.mysql-sink.channel = mysql-channel 
# sink class, must not be changed 
cygnusagent.sinks.mysql-sink.type = es.tid.fiware.fiwareconnectors.cygnus.sinks.OrionMySQLSink 
# the FQDN/IP address where the MySQL server runs 
cygnusagent.sinks.mysql-sink.mysql_host = localhost 
# the port where the MySQL server listes for incomming connections 
cygnusagent.sinks.mysql-sink.mysql_port = 3306 
# a valid user in the MySQL server 
cygnusagent.sinks.mysql-sink.mysql_username = root 
# password for the user above 
cygnusagent.sinks.mysql-sink.mysql_password = klasika 
# how the attributes are stored, either per row either per column (row, column) 
cygnusagent.sinks.mysql-sink.attr_persistence = column 

#============================================= 
# hdfs-channel configuration 
# channel type (must not be changed) 
cygnusagent.channels.hdfs-channel.type = memory 
# capacity of the channel 
cygnusagent.channels.hdfs-channel.capacity = 1000 
# amount of bytes that can be sent per transaction 
cygnusagent.channels.hdfs-channel.transactionCapacity = 100 

#============================================= 
# ckan-channel configuration 
# channel type (must not be changed) 
cygnusagent.channels.ckan-channel.type = memory 
# capacity of the channel 
cygnusagent.channels.ckan-channel.capacity = 1000 
# amount of bytes that can be sent per transaction 
cygnusagent.channels.ckan-channel.transactionCapacity = 100 

#============================================= 
# mysql-channel configuration 
# channel type (must not be changed) 
cygnusagent.channels.mysql-channel.type = memory 
# capacity of the channel 
cygnusagent.channels.mysql-channel.capacity = 1000 
# amount of bytes that can be sent per transaction 
cygnusagent.channels.mysql-channel.transactionCapacity = 100 

Хотя я не использую OrionHDFSSink и OrionCKANSink, я не трогайте те конфигурации, потому что я на самом деле не уверенная погода.

Когда я, наконец, subscribeContext и целевой Cygnus @ По умолчанию используется порт 5050, я получаю нормальный ответ, но ничего не создается в моей базе данных

Что я здесь делаю неправильно?

Answer 1

Прежде всего, не стесняйтесь удалить детали конфигурации HDFS и CKAN. Во время работы Cygnus вы избежите ненужных журналов, связанных с этими компонентами. Конечно, не забудьте удалить все ссылки на приемники и каналы; а именно:

cygnusagent.sources = http-source 
cygnusagent.sinks = mysql-sink 
cygnusagent.channels = mysql-channel 
... 
cygnusagent.sources.http-source.channels = mysql-channel 

Во-вторых, ответ на ваш вопрос можно найти в документации:

В таблицах, можно найти два варианта:

• Фиксированные строки 8-поля, как обычно: recvTimeTs, recvTime, entityId, entityType, attrName, attrType, attrValue и attrMd. Эти таблицы (и базы данных) создаются во время выполнения, если таблица не существовала ранее до вставки строки. Что касается attrValue, то в его простейшей форме это значение является просто строкой, но поскольку Orion 0.11.0 это может быть объект Json или массив Json. Что касается attrMd, он содержит строковую сериализацию массива метаданных для атрибута в Json (если атрибут не содержит метаданных, вставлен пустой массив []),
• Два столбца на каждый атрибут объекта (один для значения и другое для метаданных), плюс столбец добавления о времени приема данных (recv_time). Этот тип таблиц (и баз данных) должен быть предварительно предоставлен для выполнения Cygnus, поскольку каждый объект может иметь различное количество атрибутов, а уведомления должны обеспечивать уведомление о значении для каждого атрибута.

Поведение разъема относительно внутреннего представления данных определяется с помощью параметра конфигурации attr_persistence, значениями которого могут быть строки или столбцы.

Может быть, есть проблема с Writting, я думаю, что последний пункт должен ENDA в «... целые значения могут быть строки или столбца, и чье поведение соответствуют параметрам, описанным выше, соответственно».

I.e. если вы используете режим colummn, тогда база данных и таблицы должны быть подготовлены в расширенном режиме.

Существует similar question, где я более подробно объясню такое поведение.

HTH!