diff --git a/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml b/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml index b01ab7c0fa..8cfad035c3 100644 --- a/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml +++ b/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml @@ -16,5 +16,8 @@ version: "3.4" services: iotdb: + # 0.13.3 is the only recommended production ready version for edge side (standalone) deployment + image: apache/iotdb:0.13.3-node ports: - - 6667:6667 + # rpc port + - "6667:6667" diff --git a/installer/cli/deploy/standalone/iotdb/docker-compose.yml b/installer/cli/deploy/standalone/iotdb/docker-compose.yml index eb315f3433..f45807b9d6 100644 --- a/installer/cli/deploy/standalone/iotdb/docker-compose.yml +++ b/installer/cli/deploy/standalone/iotdb/docker-compose.yml @@ -16,9 +16,10 @@ version: "3.4" services: iotdb: - image: apache/iotdb + # 0.13.3 is the only recommended production ready version for edge side (standalone) deployment + image: apache/iotdb:0.13.3-node volumes: - - iotdb:/iotdb/data + - iotdb:/iotdb logging: driver: "json-file" options: diff --git a/pom.xml b/pom.xml index e70ec21ed9..045286e085 100644 --- a/pom.xml +++ b/pom.xml @@ -157,7 +157,6 @@ 2.2.0 0.10.0 22.3.1 - 1.0.0 1.5.0 2.1.0 2.3.2 @@ -1283,17 +1282,6 @@ flink-runtime_2.11 ${flink.version} - - org.apache.iotdb - iotdb-jdbc - ${iotdb.version} - - - javax.servlet - servlet-api - - - org.apache.opennlp opennlp-tools diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml index 0cea72e1df..16128fd3d6 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml @@ -55,7 +55,10 @@ org.apache.iotdb - iotdb-jdbc + iotdb-session + + + 0.13.3 org.eclipse.milo @@ -108,6 +111,18 @@ bcpkix-jdk15on + + + junit + junit + test + + + org.mockito + mockito-core + test + + javax.xml.bind diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java index 7e023167fb..df8300f710 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java @@ -18,166 +18,112 @@ package org.apache.streampipes.sinks.databases.jvm.iotdb; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.logging.api.Logger; import org.apache.streampipes.model.runtime.Event; -import org.apache.streampipes.model.schema.EventProperty; -import org.apache.streampipes.model.schema.EventPropertyPrimitive; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.JdbcClient; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.DbDataTypeFactory; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.DbDataTypes; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.ParameterInformation; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.SupportedDbEngines; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.utils.SQLStatementUtils; +import org.apache.streampipes.model.runtime.field.AbstractField; import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext; import org.apache.streampipes.wrapper.runtime.EventSink; -import java.sql.SQLException; -import java.sql.Statement; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.pool.SessionPool; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.utils.Binary; -public class IotDb extends JdbcClient implements EventSink { +import java.util.ArrayList; +import java.util.List; +import java.util.Map; - private IotDbParameters params; - private static Logger log; +public class IotDb implements EventSink { - private String timestampField; + private static Logger logger; - private final SupportedDbEngines dbEngine = SupportedDbEngines.IOT_DB; + private String timestampFieldId; + private String deviceId; + + // IoTDB provides a connection pool (SessionPool) for Native API. + // Using the interface, we need to define the pool size. + // If we can not get a session connection in 60 seconds, there is a warning log but the program will hang. + // If a session has finished an operation, it will be put back to the pool automatically. + // If a session connection is broken, the session will be removed automatically and the pool will try to create a + // new session and redo the operation. + private SessionPool sessionPool; @Override - public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext) - throws SpRuntimeException { - - this.params = parameters; - log = parameters.getGraph().getLogger(IotDb.class); - timestampField = parameters.getTimestampField(); - - // tablename is the identifier for the storage group in the IoTDB Adapter (e.g. root.data.table1) in which all - // time series are written - //TODO: Add better regular expression - initializeJdbc( - parameters.getGraph().getInputStreams().get(0).getEventSchema(), - parameters, - dbEngine, - log); + public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext) { + logger = parameters.getGraph().getLogger(IotDb.class); + + deviceId = "root." + parameters.getDatabase() + "." + parameters.getDevice(); + timestampFieldId = parameters.getTimestampField(); + // In our case, the pool size is set to 2. + // One connection is for current requests, and the other is a backup for fast-recovery when connection dies. + sessionPool = new SessionPool.Builder().maxSize(2).enableCompression(false).host(parameters.getHost()) + .port(parameters.getPort()).user(parameters.getUser()).password(parameters.getPassword()).build(); } @Override public void onEvent(Event event) { - try { - if (event.getRaw().containsKey("value")) { - // Renaming value. Very ugly - event.addField("value_1", event.getFieldBySelector("s0::value").getRawValue()); - event.removeFieldBySelector("s0::value"); - } - save(event); - } catch (SpRuntimeException e) { - log.error(e.getMessage()); + if (event == null) { + return; } - } - @Override - public void onDetach() throws SpRuntimeException { - closeAll(); - } - - @Override - protected void save(final Event event) throws SpRuntimeException { - checkConnected(); - try { - Long timestampValue = event.getFieldBySelector(timestampField).getAsPrimitive().getAsLong(); - event.removeFieldBySelector(timestampField); - Statement statement; - statement = connection.createStatement(); - StringBuilder sb1 = new StringBuilder(); - StringBuilder sb2 = new StringBuilder(); - //TODO: Check for SQL-Injection - // Timestamp must be in the beginning of the values - sb1.append("INSERT INTO ").append(this.params.getDbTable()).append("(timestamp, "); - sb2.append(" VALUES (").append(timestampValue).append(", "); - for (String s : event.getRaw().keySet()) { - sb1.append(s).append(", "); - if (event.getFieldByRuntimeName(s).getRawValue() instanceof String) { - sb2.append("\"").append(event.getFieldByRuntimeName(s).getRawValue().toString()).append("\", "); - } else { - sb2.append(event.getFieldByRuntimeName(s).getRawValue().toString()).append(", "); - } - } - sb1.setLength(sb1.length() - 2); - sb2.setLength(sb2.length() - 2); - sb1.append(") ").append(sb2).append(")"); - statement.execute(sb1.toString()); - } catch (SQLException e) { - e.printStackTrace(); + final AbstractField timestampAbstractField = event.getFieldBySelector(timestampFieldId); + final Long timestamp = timestampAbstractField.getAsPrimitive().getAsLong(); + if (timestamp == null) { + return; } - } - @Override - protected void ensureDatabaseExists(String url, String databaseName) throws SpRuntimeException { - SQLStatementUtils.checkRegEx(this.params.getDbTable(), "Storage Group name", this.dbDescription); - try { - Statement statement = connection.createStatement(); - statement.execute("SET STORAGE GROUP TO " + this.params.getDbTable()); - } catch (SQLException e) { - // Storage group already exists - //TODO: Catch other exceptions + final Map measurementValuePairs = event.getRaw(); + // should be at least a timestamp field and a measurement field + if (measurementValuePairs.size() <= 1) { + return; } - } - /** - * Needs to be reimplemented since the IoTDB JDBC implementation does not support the methods used in the - * JDBC-Client class - * - * @param url The JDBC url containing the needed information (e.g. "jdbc:iotdb://127.0.0.1:6667/") - * @throws SpRuntimeException - */ - @Override - protected void ensureTableExists(String url, String databaseName) throws SpRuntimeException { - int index = 1; - this.statementHandler.putEventParameterMap("timestamp", - new ParameterInformation(index++, DbDataTypeFactory.getLong(dbEngine))); - for (EventProperty eventProperty : this.tableDescription.getEventSchema().getEventProperties()) { - try { - if (eventProperty.getRuntimeName().equals(timestampField.substring(4))) { - continue; - } - Statement statement = null; - statement = connection.createStatement(); - // The identifier cannot be called "value" - //TODO: Do not simply add a _1 but look instead, if the name is already taken - String runtimeName = eventProperty.getRuntimeName(); - if (eventProperty.getRuntimeName().equals("value")) { - runtimeName = "value_1"; - } - DbDataTypes datatype = extractAndAddEventPropertyRuntimeType(eventProperty, index++); - - statement.execute("CREATE TIMESERIES " - + params.getDbTable() - + "." - + runtimeName - + " WITH DATATYPE=" - + datatype.toString() - + ", ENCODING=PLAIN"); - } catch (SQLException e) { - // Probably because it already exists - //TODO: Add better exception handling - e.printStackTrace(); + final int measurementFieldCount = measurementValuePairs.size() - 1; + final List measurements = new ArrayList<>(measurementFieldCount); + final List types = new ArrayList<>(measurementFieldCount); + final List values = new ArrayList<>(measurementFieldCount); + + for (Map.Entry measurementValuePair : measurementValuePairs.entrySet()) { + if (timestampAbstractField.getFieldNameIn().equals(measurementValuePair.getKey())) { + continue; + } + + measurements.add(measurementValuePair.getKey()); + + final Object value = measurementValuePair.getValue(); + if (value instanceof Integer) { + types.add(TSDataType.INT32); + values.add(value); + } else if (value instanceof Long) { + types.add(TSDataType.INT64); + values.add(value); + } else if (value instanceof Float) { + types.add(TSDataType.FLOAT); + values.add(value); + } else if (value instanceof Double) { + types.add(TSDataType.DOUBLE); + values.add(value); + } else if (value instanceof Boolean) { + types.add(TSDataType.BOOLEAN); + values.add(value); + } else { + types.add(TSDataType.TEXT); + values.add(Binary.valueOf(value.toString())); } } - //tableExists = true; - } - private DbDataTypes extractAndAddEventPropertyRuntimeType(EventProperty eventProperty, int index) { - // Supported datatypes can be found here: https://iotdb.apache.org/#/Documents/0.8.0/chap2/sec2 - DbDataTypes dataType = DbDataTypes.TEXT; - if (eventProperty instanceof EventPropertyPrimitive) { - dataType = DbDataTypeFactory.getFromUri(((EventPropertyPrimitive) eventProperty).getRuntimeType(), - SupportedDbEngines.IOT_DB); - this.statementHandler.putEventParameterMap(eventProperty.getRuntimeName(), - new ParameterInformation(index, dataType)); + try { + sessionPool.insertRecord(deviceId, timestamp, measurements, types, values); + } catch (IoTDBConnectionException | StatementExecutionException e) { + logger.error("Failed to save event to IoTDB, because: " + e.getMessage()); + e.printStackTrace(); } + } - return dataType; + @Override + public void onDetach() { + sessionPool.close(); } } diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java index 6731397a4f..7bc974bbac 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java @@ -34,52 +34,46 @@ public class IotDbController extends StandaloneEventSinkDeclarer { - private static final String DATABASE_HOST_KEY = "db_host"; - private static final String DATABASE_PORT_KEY = "db_port"; - private static final String STORAGE_GROUP_KEY = "db_storage_group"; - private static final String DATABASE_USER_KEY = "db_user"; - private static final String DATABASE_PASSWORD_KEY = "db_password"; - private static final String TIMESTAMPE_MAPPING_KEY = "timestamp_mapping"; + private static final String HOST_KEY = "db_host"; + private static final String PORT_KEY = "db_port"; + + private static final String DATABASE_KEY = "db_database"; + private static final String DEVICE_KEY = "db_device"; + + private static final String USER_KEY = "db_user"; + private static final String PASSWORD_KEY = "db_password"; + + private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping"; @Override public DataSinkDescription declareModel() { - return DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb") - .withLocales(Locales.EN) - .withAssets(Assets.DOCUMENTATION, Assets.ICON) - .category(DataSinkType.DATABASE) - .requiredStream(StreamRequirementsBuilder.create() - .requiredPropertyWithUnaryMapping( - EpRequirements.timestampReq(), - Labels.withId(TIMESTAMPE_MAPPING_KEY), - PropertyScope.NONE) - .build()) - .requiredTextParameter(Labels.withId(DATABASE_HOST_KEY)) - .requiredIntegerParameter(Labels.withId(DATABASE_PORT_KEY), 6667) - .requiredTextParameter(Labels.withId(STORAGE_GROUP_KEY)) - .requiredTextParameter(Labels.withId(DATABASE_USER_KEY)) - .requiredSecret(Labels.withId(DATABASE_PASSWORD_KEY)) - .build(); + return DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb").withLocales(Locales.EN) + .withAssets(Assets.DOCUMENTATION, Assets.ICON).category(DataSinkType.DATABASE).requiredStream( + StreamRequirementsBuilder.create() + .requiredPropertyWithUnaryMapping(EpRequirements.timestampReq(), Labels.withId(TIMESTAMP_MAPPING_KEY), + PropertyScope.NONE).build()).requiredTextParameter(Labels.withId(HOST_KEY)) + .requiredIntegerParameter(Labels.withId(PORT_KEY), 6667).requiredTextParameter(Labels.withId(USER_KEY), "root") + .requiredSecret(Labels.withId(PASSWORD_KEY)).requiredTextParameter(Labels.withId(DATABASE_KEY)) + .requiredTextParameter(Labels.withId(DEVICE_KEY)).build(); } @Override public ConfiguredEventSink onInvocation(DataSinkInvocation graph, DataSinkParameterExtractor extractor) { - String hostname = extractor.singleValueParameter(DATABASE_HOST_KEY, String.class); - Integer port = extractor.singleValueParameter(DATABASE_PORT_KEY, Integer.class); - String dbStorageGroup = extractor.singleValueParameter(STORAGE_GROUP_KEY, String.class); - String user = extractor.singleValueParameter(DATABASE_USER_KEY, String.class); - String password = extractor.secretValue(DATABASE_PASSWORD_KEY); - String timestampField = extractor.mappingPropertyValue(TIMESTAMPE_MAPPING_KEY); + final String host = extractor.singleValueParameter(HOST_KEY, String.class); + final Integer port = extractor.singleValueParameter(PORT_KEY, Integer.class); + + final String database = extractor.singleValueParameter(DATABASE_KEY, String.class); + final String device = extractor.singleValueParameter(DEVICE_KEY, String.class); + + final String user = extractor.singleValueParameter(USER_KEY, String.class); + final String password = extractor.secretValue(PASSWORD_KEY); + + final String timestampField = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY); - IotDbParameters params = new IotDbParameters(graph, - hostname, - port, - dbStorageGroup, - user, - password, - false, // SSL connection not yet implemented for IoT DB - timestampField); + final IotDbParameters params = + new IotDbParameters(graph, host, port, user, password, database, device, timestampField); return new ConfiguredEventSink<>(params, IotDb::new); } diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java index 3150717244..4f41610c8e 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java @@ -19,36 +19,68 @@ package org.apache.streampipes.sinks.databases.jvm.iotdb; import org.apache.streampipes.model.graph.DataSinkInvocation; -import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.JdbcConnectionParameters; +import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams; -public class IotDbParameters extends JdbcConnectionParameters { +public class IotDbParameters extends EventSinkBindingParams { - private String timestampField; + private final String host; + private final Integer port; + + private final String user; + private final String password; + + private final String database; + // entity + private final String device; + + private final String timestampField; public IotDbParameters(DataSinkInvocation graph, - String iotDbHost, - Integer iotDbPort, - String dbStorageGroup, + String host, + Integer port, String user, String password, - boolean sslEnabled, + String database, + String device, String timestampField) { - super( - graph, - iotDbHost, - iotDbPort, - "", - user, - password, - "root." + dbStorageGroup, - sslEnabled, - null, - false - ); + super(graph); + + this.host = host; + this.port = port; + + this.user = user; + this.password = password; + + this.database = database; + this.device = device; this.timestampField = timestampField; } + public String getHost() { + return host; + } + + public Integer getPort() { + return port; + } + + public String getUser() { + return user; + } + + public String getPassword() { + return password; + } + + public String getDatabase() { + return database; + } + + public String getDevice() { + return device; + } + public String getTimestampField() { return timestampField; } diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java index 14887d3bec..206535b6dd 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java @@ -29,8 +29,6 @@ public static DbDataTypes getInteger(SupportedDbEngines sqlEngine) throws SpRunt switch (sqlEngine) { case MY_SQL: return DbDataTypes.INT; - case IOT_DB: - return DbDataTypes.INT32; case POSTGRESQL: return DbDataTypes.INTEGER; } @@ -40,8 +38,6 @@ public static DbDataTypes getInteger(SupportedDbEngines sqlEngine) throws SpRunt public static DbDataTypes getLong(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: - return DbDataTypes.INT64; case MY_SQL: case POSTGRESQL: return DbDataTypes.BIGINT; @@ -52,7 +48,6 @@ public static DbDataTypes getLong(SupportedDbEngines sqlEngine) throws SpRuntime public static DbDataTypes getFloat(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: case MY_SQL: return DbDataTypes.FLOAT; case POSTGRESQL: @@ -63,8 +58,6 @@ public static DbDataTypes getFloat(SupportedDbEngines sqlEngine) throws SpRuntim public static DbDataTypes getDouble(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: - return DbDataTypes.DOUBLE; case MY_SQL: case POSTGRESQL: return DbDataTypes.DOUBLE_PRECISION; @@ -74,8 +67,6 @@ public static DbDataTypes getDouble(SupportedDbEngines sqlEngine) throws SpRunti public static DbDataTypes getShortString(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: - return DbDataTypes.TEXT; case MY_SQL: case POSTGRESQL: return DbDataTypes.VAR_CHAR; @@ -85,7 +76,6 @@ public static DbDataTypes getShortString(SupportedDbEngines sqlEngine) throws Sp public static DbDataTypes getLongString(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: case MY_SQL: case POSTGRESQL: return DbDataTypes.TEXT; @@ -95,7 +85,6 @@ public static DbDataTypes getLongString(SupportedDbEngines sqlEngine) throws SpR public static DbDataTypes getBoolean(SupportedDbEngines sqlEngine) throws SpRuntimeException { switch (sqlEngine) { - case IOT_DB: case MY_SQL: case POSTGRESQL: return DbDataTypes.BOOLEAN; @@ -108,8 +97,6 @@ public static DbDataTypes getTimestamp(SupportedDbEngines sqlEngine) throws SpRu case MY_SQL: case POSTGRESQL: return DbDataTypes.TIMESTAMP; - case IOT_DB: - return getLong(sqlEngine); } throw new SpRuntimeException("Database engine " + sqlEngine + " does not support timestamps."); } @@ -119,7 +106,6 @@ public static DbDataTypes getDate(SupportedDbEngines sqlEngine) throws SpRuntime case MY_SQL: case POSTGRESQL: return DbDataTypes.DATE; - case IOT_DB: } throw new SpRuntimeException("Database engine " + sqlEngine + " does not support dates."); } @@ -129,7 +115,6 @@ public static DbDataTypes getTime(SupportedDbEngines sqlEngine) throws SpRuntime case MY_SQL: case POSTGRESQL: return DbDataTypes.TIME; - case IOT_DB: } throw new SpRuntimeException("Database engine " + sqlEngine + " does not support time."); } @@ -138,7 +123,6 @@ public static DbDataTypes getDatetime(SupportedDbEngines sqlEngine) throws SpRun switch (sqlEngine) { case MY_SQL: return DbDataTypes.DATETIME; - case IOT_DB: case POSTGRESQL: } throw new SpRuntimeException("Database engine " + sqlEngine + " does not support datetime."); @@ -189,17 +173,14 @@ public static DbDataTypes getFromUri(final String uri, SupportedDbEngines sqlEng public static Datatypes getDataType(DbDataTypes dbDataType) throws SpRuntimeException { switch (dbDataType) { - case BOOL: case BOOLEAN: return Datatypes.Boolean; case TEXT: case VAR_CHAR: - case STRING: case TIMESTAMP: case DATE: case TIME: return Datatypes.String; - case DOUBLE: case DOUBLE_PRECISION: return Datatypes.Double; case FLOAT: @@ -208,10 +189,8 @@ public static Datatypes getDataType(DbDataTypes dbDataType) throws SpRuntimeExce return Datatypes.Float; case BIGINT: case TINYINT: - case INT64: return Datatypes.Long; case INT: - case INT32: case INTEGER: return Datatypes.Integer; default: diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java index 5691a31f43..25b0ecf6a3 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java @@ -21,19 +21,14 @@ public enum DbDataTypes { - BOOL("BOOL"), BOOLEAN("BOOLEAN"), TEXT("TEXT"), VAR_CHAR("VARCHAR255"), - STRING("STRING"), - DOUBLE("DOUBLE"), DOUBLE_PRECISION("DOUBLE PRECISION"), FLOAT("FLOAT"), REAL("REAL"), BIGINT("BIGINT"), TINYINT("TINYINT"), - INT64("INT64"), - INT32("INT32"), INT("INT"), INTEGER("INTEGER"), TIMESTAMP("TIMESTAMP"), diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java index 21d6176537..55a0a191d3 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java @@ -20,7 +20,6 @@ public enum SupportedDbEngines { MY_SQL("mysql", "com.mysql.cj.jdbc.Driver", ".*"), - IOT_DB("iotdb", "org.apache.iotdb.jdbc.IoTDBDriver", ".*"), POSTGRESQL("postgresql", "org.postgresql.Driver", "^[a-zA-Z_][a-zA-Z0-9_]*$"); private final String urlName; diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md index d02b0d429d..eacab0a2e0 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md @@ -28,6 +28,10 @@ Stores events in a IoTDB database. +Events will be stored in timeseries `root.${Database Name}.${Device (Entity) Name}.${Event Key}`. + +Please reference to [https://iotdb.apache.org/](https://iotdb.apache.org/) for more information. + *** ## Required input @@ -46,11 +50,6 @@ The hostname of the IoTDB instance. The port of the IoTDB instance (default 6667). -### Storage Group Name - -The name of the storage group where events will be stored (will be created if it does not exist). -For each element of the stream a new time series will be created. - ### Username The username for the IoTDB Server. @@ -59,6 +58,28 @@ The username for the IoTDB Server. The password for the IoTDB Server. +### **Database Name** + +The name of the database where events will be stored (will be created if it does not exist). + +A database is a group of devices (entities). Users can create any prefix path as a database. + +### **Device (Entity) Name** + +The name of the device (entity) where events will be stored. + +A device (also called entity) is an equipped with measurements in real scenarios. In IoTDB, all measurements should have +their corresponding devices. + +### **Measurements** + +All keys of fields in an event will be automatically converted to suffixes of timeseries. + +A measurement is information measured by detection equipment in an actual scene and can transform the sensed information +into an electrical signal or other desired form of information output and send it to IoTDB. + +In IoTDB, all data and paths stored are organized in units of measurements. + ## Output (not applicable for data sinks) diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en index 107b64b02f..cb16982637 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en @@ -15,8 +15,8 @@ # limitations under the License. # -org.apache.streampipes.sinks.databases.jvm.iotdb.title=IoTDB -org.apache.streampipes.sinks.databases.jvm.iotdb.description=Stores events in a IoTDB database. +org.apache.streampipes.sinks.databases.jvm.iotdb.title=Apache IoTDB +org.apache.streampipes.sinks.databases.jvm.iotdb.description=Stores events in a IoTDB database. Events will be stored in timeseries root.${Database Name}.${Device (Entity) Name}.${Event Key}. db_host.title=Hostname db_host.description=The hostname of the IoTDB instance @@ -24,14 +24,17 @@ db_host.description=The hostname of the IoTDB instance db_port.title=Port db_port.description=The port of the IoTDB instance (default 6667) -db_storage_group.title=Storage Group Name -db_storage_group.description=The name of the storage group where events will be stored +db_database.title=Database Name +db_database.description=The name of the database where events will be stored. A database is a group of devices (entities). Users can create any prefix path as a database. + +db_device.title=Device (Entity) Name +db_device.description=The name of the device (entity) where events will be stored. A device (also called entity) is an equipped with measurements in real scenarios. In IoTDB, all measurements should have their corresponding devices. db_user.title=Username -db_user.description=The username for the IoTDB Server +db_user.description=The username for the IoTDB Server (default "root") db_password.title=Password -db_password.description=The password for the IoTDB Server +db_password.description=The password for the IoTDB Server (default "root") timestamp_mapping.title=Timestamp timestamp_mapping.description=The timestamp of the event diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java new file mode 100644 index 0000000000..1bf3092529 --- /dev/null +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.sinks.databases.jvm.iotdb; + +import org.apache.streampipes.model.graph.DataSinkInvocation; + +import org.junit.Assert; +import org.junit.Test; + +import static org.mockito.Mockito.mock; + +public class TestIotDbParameters { + + @Test + public void testIotDbParameters() { + final String host = "localhost"; + final Integer port = 6667; + final String user = "user"; + final String password = "password"; + final String database = "database"; + final String device = "device"; + final String timestampField = "timestampField"; + + final IotDbParameters parameters = + new IotDbParameters(mock(DataSinkInvocation.class), host, port, user, password, database, device, + timestampField); + + Assert.assertEquals(host, parameters.getHost()); + Assert.assertEquals(port, parameters.getPort()); + Assert.assertEquals(user, parameters.getUser()); + Assert.assertEquals(password, parameters.getPassword()); + Assert.assertEquals(database, parameters.getDatabase()); + Assert.assertEquals(device, parameters.getDevice()); + Assert.assertEquals(timestampField, parameters.getTimestampField()); + } +} \ No newline at end of file