tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge branch 'lts-4.3' into lts-4.4

pull/16111/head
Viacheslav Klimov 3 days ago
parent
88ac3b58de 25497c2366
commit
e4e4650828
6 changed files with 0 additions and 724 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 74
      tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java
  2. 102
      tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java
  3. 282
      tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java
  4. 92
      tools/src/main/java/org/thingsboard/client/tools/migrator/README.md
  5. 88
      tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java
  6. 86
      tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java

74
tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java
View File

@ -1,74 +0,0 @@
/**
* Copyright © 2016-2026 The Thingsboard Authors
*
* Licensed 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.thingsboard.client.tools.migrator;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.thingsboard.server.common.data.StringUtils;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class DictionaryParser {
private Map<String, String> dictionaryParsed = new HashMap<>();
public DictionaryParser(File sourceFile) throws IOException {
parseDictionaryDump(FileUtils.lineIterator(sourceFile));
}
public String getKeyByKeyId(String keyId) {
return dictionaryParsed.get(keyId);
}
private boolean isBlockFinished(String line) {
return StringUtils.isBlank(line) || line.equals("\\.");
}
private boolean isBlockStarted(String line) {
return line.startsWith("COPY public.key_dictionary (");
}
private void parseDictionaryDump(LineIterator iterator) throws IOException {
try {
String tempLine;
while (iterator.hasNext()) {
tempLine = iterator.nextLine();
if (isBlockStarted(tempLine)) {
processBlock(iterator);
}
}
} finally {
iterator.close();
}
}
private void processBlock(LineIterator lineIterator) {
String tempLine;
String[] lineSplited;
while(lineIterator.hasNext()) {
tempLine = lineIterator.nextLine();
if(isBlockFinished(tempLine)) {
return;
}
lineSplited = tempLine.split("\t");
dictionaryParsed.put(lineSplited[1], lineSplited[0]);
}
}
}

102
tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java
View File

@ -1,102 +0,0 @@
/**
* Copyright © 2016-2026 The Thingsboard Authors
*
* Licensed 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.thingsboard.client.tools.migrator;
import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import java.io.File;
public class MigratorTool {
public static void main(String[] args) {
CommandLine cmd = parseArgs(args);
try {
boolean castEnable = Boolean.parseBoolean(cmd.getOptionValue("castEnable"));
File allTelemetrySource = new File(cmd.getOptionValue("telemetryFrom"));
File tsSaveDir = null;
File partitionsSaveDir = null;
File latestSaveDir = null;
RelatedEntitiesParser allEntityIdsAndTypes =
new RelatedEntitiesParser(new File(cmd.getOptionValue("relatedEntities")));
DictionaryParser dictionaryParser = new DictionaryParser(allTelemetrySource);
if(cmd.getOptionValue("latestTelemetryOut") != null) {
latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut"));
}
if(cmd.getOptionValue("telemetryOut") != null) {
tsSaveDir = new File(cmd.getOptionValue("telemetryOut"));
partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut"));
}
new PgCaMigrator(allTelemetrySource, tsSaveDir, partitionsSaveDir, latestSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable).migrate();
} catch (Throwable th) {
th.printStackTrace();
throw new IllegalStateException("failed", th);
}
}
private static CommandLine parseArgs(String[] args) {
Options options = new Options();
Option telemetryAllFrom = new Option("telemetryFrom", "telemetryFrom", true, "telemetry source file");
telemetryAllFrom.setRequired(true);
options.addOption(telemetryAllFrom);
Option latestTsOutOpt = new Option("latestOut", "latestTelemetryOut", true, "latest telemetry save dir");
latestTsOutOpt.setRequired(false);
options.addOption(latestTsOutOpt);
Option tsOutOpt = new Option("tsOut", "telemetryOut", true, "sstable save dir");
tsOutOpt.setRequired(false);
options.addOption(tsOutOpt);
Option partitionOutOpt = new Option("partitionsOut", "partitionsOut", true, "partitions save dir");
partitionOutOpt.setRequired(false);
options.addOption(partitionOutOpt);
Option castOpt = new Option("castEnable", "castEnable", true, "cast String to Double if possible");
castOpt.setRequired(true);
options.addOption(castOpt);
Option relatedOpt = new Option("relatedEntities", "relatedEntities", true, "related entities source file path");
relatedOpt.setRequired(true);
options.addOption(relatedOpt);
HelpFormatter formatter = new HelpFormatter();
CommandLineParser parser = new BasicParser();
try {
return parser.parse(options, args);
} catch (ParseException e) {
System.out.println(e.getMessage());
formatter.printHelp("utility-name", options);
System.exit(1);
}
return null;
}
}

282
tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java
View File

@ -1,282 +0,0 @@
/**
* Copyright © 2016-2026 The Thingsboard Authors
*
* Licensed 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.thingsboard.client.tools.migrator;
import com.google.common.collect.Lists;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.server.common.data.StringUtils;
import java.io.File;
import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Collectors;
public class PgCaMigrator {
private final long LOG_BATCH = 1000000;
private final long rowPerFile = 1000000;
private long linesTsMigrated = 0;
private long linesLatestMigrated = 0;
private long castErrors = 0;
private long castedOk = 0;
private long currentWriterCount = 1;
private final File sourceFile;
private final boolean castStringIfPossible;
private final RelatedEntitiesParser entityIdsAndTypes;
private final DictionaryParser keyParser;
private CQLSSTableWriter currentTsWriter;
private CQLSSTableWriter currentPartitionsWriter;
private CQLSSTableWriter currentTsLatestWriter;
private final Set<String> partitions = new HashSet<>();
private File outTsDir;
private File outTsLatestDir;
public PgCaMigrator(File sourceFile,
File ourTsDir,
File outTsPartitionDir,
File outTsLatestDir,
RelatedEntitiesParser allEntityIdsAndTypes,
DictionaryParser dictionaryParser,
boolean castStringsIfPossible) {
this.sourceFile = sourceFile;
this.entityIdsAndTypes = allEntityIdsAndTypes;
this.keyParser = dictionaryParser;
this.castStringIfPossible = castStringsIfPossible;
if(outTsLatestDir != null) {
this.currentTsLatestWriter = WriterBuilder.getLatestWriter(outTsLatestDir);
this.outTsLatestDir = outTsLatestDir;
}
if(ourTsDir != null) {
this.currentTsWriter = WriterBuilder.getTsWriter(ourTsDir);
this.currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir);
this.outTsDir = ourTsDir;
}
}
public void migrate() throws IOException {
boolean isTsDone = false;
boolean isLatestDone = false;
String line;
LineIterator iterator = FileUtils.lineIterator(this.sourceFile);
try {
while(iterator.hasNext()) {
line = iterator.nextLine();
if(!isLatestDone && isBlockLatestStarted(line)) {
System.out.println("START TO MIGRATE LATEST");
long start = System.currentTimeMillis();
processBlock(iterator, currentTsLatestWriter, outTsLatestDir, this::toValuesLatest);
System.out.println("TOTAL LINES MIGRATED: " + linesLatestMigrated + ", FORMING OF SSL FOR LATEST TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms.");
isLatestDone = true;
}
if(!isTsDone && isBlockTsStarted(line)) {
System.out.println("START TO MIGRATE TS");
long start = System.currentTimeMillis();
processBlock(iterator, currentTsWriter, outTsDir, this::toValuesTs);
System.out.println("TOTAL LINES MIGRATED: " + linesTsMigrated + ", FORMING OF SSL FOR TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms.");
isTsDone = true;
}
}
System.out.println("Partitions collected " + partitions.size());
long startTs = System.currentTimeMillis();
for (String partition : partitions) {
String[] split = partition.split("\\|");
List<Object> values = Lists.newArrayList();
values.add(split[0]);
values.add(UUID.fromString(split[1]));
values.add(split[2]);
values.add(Long.parseLong(split[3]));
currentPartitionsWriter.addRow(values);
}
System.out.println(new Date() + " Migrated partitions " + partitions.size() + " in " + (System.currentTimeMillis() - startTs));
System.out.println();
System.out.println("Finished migrate Telemetry");
} finally {
iterator.close();
currentTsLatestWriter.close();
currentTsWriter.close();
currentPartitionsWriter.close();
}
}
private void logLinesProcessed(long lines) {
if (lines % LOG_BATCH == 0) {
System.out.println(new Date() + " lines processed = " + lines + " in, castOk " + castedOk + " castErr " + castErrors);
}
}
private void logLinesMigrated(long lines) {
if(lines % LOG_BATCH == 0) {
System.out.println(new Date() + " lines migrated = " + lines + " in, castOk " + castedOk + " castErr " + castErrors);
}
}
private void addTypeIdKey(List<Object> result, List<String> raw) {
result.add(entityIdsAndTypes.getEntityType(raw.get(0)));
result.add(UUID.fromString(raw.get(0)));
result.add(keyParser.getKeyByKeyId(raw.get(1)));
}
private void addPartitions(List<Object> result, List<String> raw) {
long ts = Long.parseLong(raw.get(2));
long partition = toPartitionTs(ts);
result.add(partition);
result.add(ts);
}
private void addTimeseries(List<Object> result, List<String> raw) {
result.add(Long.parseLong(raw.get(2)));
}
private void addValues(List<Object> result, List<String> raw) {
result.add(raw.get(3).equals("\\N") ? null : raw.get(3).equals("t") ? Boolean.TRUE : Boolean.FALSE);
result.add(raw.get(4).equals("\\N") ? null : raw.get(4));
result.add(raw.get(5).equals("\\N") ? null : Long.parseLong(raw.get(5)));
result.add(raw.get(6).equals("\\N") ? null : Double.parseDouble(raw.get(6)));
result.add(raw.get(7).equals("\\N") ? null : raw.get(7));
}
private List<Object> toValuesTs(List<String> raw) {
logLinesMigrated(linesTsMigrated++);
List<Object> result = new ArrayList<>();
addTypeIdKey(result, raw);
addPartitions(result, raw);
addValues(result, raw);
processPartitions(result);
return result;
}
private List<Object> toValuesLatest(List<String> raw) {
logLinesMigrated(linesLatestMigrated++);
List<Object> result = new ArrayList<>();
addTypeIdKey(result, raw);
addTimeseries(result, raw);
addValues(result, raw);
return result;
}
private long toPartitionTs(long ts) {
LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
return time.truncatedTo(ChronoUnit.DAYS).withDayOfMonth(1).toInstant(ZoneOffset.UTC).toEpochMilli();
}
private void processPartitions(List<Object> values) {
String key = values.get(0) + "|" + values.get(1) + "|" + values.get(2) + "|" + values.get(3);
partitions.add(key);
}
private void processBlock(LineIterator iterator, CQLSSTableWriter writer, File outDir, Function<List<String>, List<Object>> function) {
String currentLine;
long linesProcessed = 0;
while(iterator.hasNext()) {
logLinesProcessed(linesProcessed++);
currentLine = iterator.nextLine();
if(isBlockFinished(currentLine)) {
return;
}
try {
List<String> raw = Arrays.stream(currentLine.trim().split("\t"))
.map(String::trim)
.collect(Collectors.toList());
List<Object> values = function.apply(raw);
if (this.currentWriterCount == 0) {
System.out.println(new Date() + " close writer " + new Date());
writer.close();
writer = WriterBuilder.getLatestWriter(outDir);
}
if (this.castStringIfPossible) {
writer.addRow(castToNumericIfPossible(values));
} else {
writer.addRow(values);
}
currentWriterCount++;
if (currentWriterCount >= rowPerFile) {
currentWriterCount = 0;
}
} catch (Exception ex) {
System.out.println(ex.getMessage() + " -> " + currentLine);
}
}
}
private List<Object> castToNumericIfPossible(List<Object> values) {
try {
if (values.get(6) != null && NumberUtils.isCreatable(values.get(6).toString())) {
Double casted = NumberUtils.createDouble(values.get(6).toString());
List<Object> numeric = Lists.newArrayList();
numeric.addAll(values);
numeric.set(6, null);
numeric.set(8, casted);
castedOk++;
return numeric;
}
} catch (Throwable th) {
castErrors++;
}
processPartitions(values);
return values;
}
private boolean isBlockFinished(String line) {
return StringUtils.isBlank(line) || line.equals("\\.");
}
private boolean isBlockTsStarted(String line) {
return line.startsWith("COPY public.ts_kv (");
}
private boolean isBlockLatestStarted(String line) {
return line.startsWith("COPY public.ts_kv_latest (");
}
}

92
tools/src/main/java/org/thingsboard/client/tools/migrator/README.md
View File

@ -1,92 +0,0 @@
# Description:
This tool used for migrating ThingsBoard into hybrid mode from Postgres.
Performance of this tool depends on disk type and instance type (mostly on CPU resources).
But in general here are few benchmarks:
1. Creating Dump of the postgres ts_kv table -> 100GB = 90 minutes
2. If postgres table has size 100GB then dump file will be about 30GB
3. Generation SSTables from dump -> 100GB = 3 hours
4. 100GB Dump file will be converted into SSTable with size about 18GB
# Tool build Instruction:
Switch to `tools` module in Command Line and execute
mvn clean compile assembly:single
It will generate single jar file with all required dependencies inside `target dir` -> `tools-2.4.1-SNAPSHOT-jar-with-dependencies.jar`.
# Prepare requred files and run Tool:
#### Dump data from the source Postgres Database
*Do not use compression if possible because Tool can only work with uncompressed file
1. Dump related tables that need to correct save telemetry
`pg_dump -h localhost -U postgres -d thingsboard -T admin_settings -T attribute_kv -T audit_log -T component_discriptor -T device_credentials -T event -T oauth2_client_registration -T oauth2_client_registration_info -T oauth2_client_registration_template -T relation -T rule_node_state tb_schema_settings -T user_credentials > related_entities.dmp`
2. Dump `ts_kv` and child:
`pg_dump -h localhost -U postgres -d thingsboard --load-via-partition-root --data-only -t ts_kv* > ts_kv_all.dmp`
3. [Optional] Move table dumps to the instance where cassandra will be hosted
#### Prepare directory structure for SSTables
Tool use 3 different directories for saving SSTables - `ts_kv_cf`, `ts_kv_latest_cf`, `ts_kv_partitions_cf`
Create 3 empty directories. For example:
/home/user/migration/ts
/home/user/migration/ts_latest
/home/user/migration/ts_partition
#### Run tool
**If you want to migrate just `ts_kv` without `ts_kv_latest` or vice versa don't use arguments (paths) for output files*
**Note: if you run this tool on remote instance - don't forget to execute this command in `screen` to avoid unexpected termination*
```
java -jar ./tools-3.2.2-SNAPSHOT-jar-with-dependencies.jar
-telemetryFrom /home/user/dump/ts_kv_all.dmp
-relatedEntities /home/user/dump/related_entities.dmp
-latestOut /home/user/migration/ts_latest
-tsOut /home/user/migration/ts
-partitionsOut /home/user/migration/ts_partition
-castEnable false
```
*Use your paths for program arguments*
Tool execution time depends on DB size, CPU resources and Disk throughput
## Adding SSTables into Cassandra
* Note that this this part works only for single node Cassandra Cluster. If you have more nodes - it is better to use `sstableloader` tool.
1. [Optional] install Cassandra on the instance
2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this files `schema-keyspace.cql`, `schema-ts.cql` and `schema-ts-latest.cql` using `source` command
3. Stop Cassandra
4. Look at `/var/lib/cassandra/data/thingsboard` and check for names of data folders
5. Copy generated SSTable files into cassandra data dir using next command:
```
sudo find /home/user/migration/ts -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_cf-0e9aaf00ee5511e9a5fa7d6f489ffd13/ \;
sudo find /home/user/migration/ts_latest -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_latest_cf-161449d0ee5511e9a5fa7d6f489ffd13/ \;
sudo find /home/user/migration/ts_partition -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_partitions_cf-12e8fa80ee5511e9a5fa7d6f489ffd13/ \;
```
*Pay attention! Data folders have similar name `ts_kv_cf-0e9aaf00ee5511e9a5fa7d6f489ffd13`, but you have to use own*
6. Start Cassandra service and trigger compaction
Trigger compactions: `nodetool compact thingsboard`
Check compaction status: `nodetool compactionstats`
## Switch Thignsboard into Hybrid Mode
Modify Thingsboard properites file `thingsboard.yml`
- DATABASE_TS_TYPE = cassandra
- TS_KV_PARTITIONING = MONTHS
# Final steps
Start Thingsboard and verify migration

88
tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java
View File

@ -1,88 +0,0 @@
/**
* Copyright © 2016-2026 The Thingsboard Authors
*
* Licensed 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.thingsboard.client.tools.migrator;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class RelatedEntitiesParser {
private final Map<String, String> allEntityIdsAndTypes = new HashMap<>();
private final Map<String, EntityType> tableNameAndEntityType = Map.ofEntries(
Map.entry("COPY public.alarm ", EntityType.ALARM),
Map.entry("COPY public.asset ", EntityType.ASSET),
Map.entry("COPY public.customer ", EntityType.CUSTOMER),
Map.entry("COPY public.dashboard ", EntityType.DASHBOARD),
Map.entry("COPY public.device ", EntityType.DEVICE),
Map.entry("COPY public.rule_chain ", EntityType.RULE_CHAIN),
Map.entry("COPY public.rule_node ", EntityType.RULE_NODE),
Map.entry("COPY public.tenant ", EntityType.TENANT),
Map.entry("COPY public.tb_user ", EntityType.USER),
Map.entry("COPY public.entity_view ", EntityType.ENTITY_VIEW),
Map.entry("COPY public.widgets_bundle ", EntityType.WIDGETS_BUNDLE),
Map.entry("COPY public.widget_type ", EntityType.WIDGET_TYPE),
Map.entry("COPY public.tenant_profile ", EntityType.TENANT_PROFILE),
Map.entry("COPY public.device_profile ", EntityType.DEVICE_PROFILE),
Map.entry("COPY public.asset_profile ", EntityType.ASSET_PROFILE),
Map.entry("COPY public.api_usage_state ", EntityType.API_USAGE_STATE)
);
public RelatedEntitiesParser(File source) throws IOException {
processAllTables(FileUtils.lineIterator(source));
}
public String getEntityType(String uuid) {
return this.allEntityIdsAndTypes.get(uuid);
}
private boolean isBlockFinished(String line) {
return StringUtils.isBlank(line) || line.equals("\\.");
}
private void processAllTables(LineIterator lineIterator) throws IOException {
String currentLine;
try {
while (lineIterator.hasNext()) {
currentLine = lineIterator.nextLine();
for(Map.Entry<String, EntityType> entry : tableNameAndEntityType.entrySet()) {
if(currentLine.startsWith(entry.getKey())) {
processBlock(lineIterator, entry.getValue());
}
}
}
} finally {
lineIterator.close();
}
}
private void processBlock(LineIterator lineIterator, EntityType entityType) {
String currentLine;
while(lineIterator.hasNext()) {
currentLine = lineIterator.nextLine();
if(isBlockFinished(currentLine)) {
return;
}
allEntityIdsAndTypes.put(currentLine.split("\t")[0], entityType.name());
}
}
}

86
tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java
View File

@ -1,86 +0,0 @@
/**
* Copyright © 2016-2026 The Thingsboard Authors
*
* Licensed 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.thingsboard.client.tools.migrator;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
import java.io.File;
public class WriterBuilder {
private static final String tsSchema = "CREATE TABLE thingsboard.ts_kv_cf (\n" +
" entity_type text, // (DEVICE, CUSTOMER, TENANT)\n" +
" entity_id timeuuid,\n" +
" key text,\n" +
" partition bigint,\n" +
" ts bigint,\n" +
" bool_v boolean,\n" +
" str_v text,\n" +
" long_v bigint,\n" +
" dbl_v double,\n" +
" json_v text,\n" +
" PRIMARY KEY (( entity_type, entity_id, key, partition ), ts)\n" +
");";
private static final String latestSchema = "CREATE TABLE IF NOT EXISTS thingsboard.ts_kv_latest_cf (\n" +
" entity_type text, // (DEVICE, CUSTOMER, TENANT)\n" +
" entity_id timeuuid,\n" +
" key text,\n" +
" ts bigint,\n" +
" bool_v boolean,\n" +
" str_v text,\n" +
" long_v bigint,\n" +
" dbl_v double,\n" +
" json_v text,\n" +
" PRIMARY KEY (( entity_type, entity_id ), key)\n" +
") WITH compaction = { 'class' : 'LeveledCompactionStrategy' };";
private static final String partitionSchema = "CREATE TABLE IF NOT EXISTS thingsboard.ts_kv_partitions_cf (\n" +
" entity_type text, // (DEVICE, CUSTOMER, TENANT)\n" +
" entity_id timeuuid,\n" +
" key text,\n" +
" partition bigint,\n" +
" PRIMARY KEY (( entity_type, entity_id, key ), partition)\n" +
") WITH CLUSTERING ORDER BY ( partition ASC )\n" +
" AND compaction = { 'class' : 'LeveledCompactionStrategy' };";
public static CQLSSTableWriter getTsWriter(File dir) {
return CQLSSTableWriter.builder()
.inDirectory(dir.getAbsolutePath())
.forTable(tsSchema)
.using("INSERT INTO thingsboard.ts_kv_cf (entity_type, entity_id, key, partition, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
.build();
}
public static CQLSSTableWriter getLatestWriter(File dir) {
return CQLSSTableWriter.builder()
.inDirectory(dir.getAbsolutePath())
.forTable(latestSchema)
.using("INSERT INTO thingsboard.ts_kv_latest_cf (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")
.build();
}
public static CQLSSTableWriter getPartitionWriter(File dir) {
return CQLSSTableWriter.builder()
.inDirectory(dir.getAbsolutePath())
.forTable(partitionSchema)
.using("INSERT INTO thingsboard.ts_kv_partitions_cf (entity_type, entity_id, key, partition) " +
"VALUES (?, ?, ?, ?)")
.build();
}
}
Write Preview
Loading…
Cancel
Save