You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2019/09/20 17:43:32 UTC
[GitHub] [nifi] tpalfy commented on a change in pull request #3611:
NIFI-6009 ScanKudu Processor
tpalfy commented on a change in pull request #3611: NIFI-6009 ScanKudu Processor
URL: https://github.com/apache/nifi/pull/3611#discussion_r326732245
##########
File path: nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/AbstractKuduProcessor.java
##########
@@ -260,4 +266,110 @@ protected Update updateRecordToKudu(KuduTable kuduTable, Record record, List<Str
return update;
}
+ private void addPredicate(KuduScanner.KuduScannerBuilder scannerBuilder, KuduTable kuduTable, String column, Object value, String comparisonOp) {
+ ColumnSchema columnSchema = kuduTable.getSchema().getColumn(column);
+ KuduPredicate predicate = KuduPredicate.newComparisonPredicate(columnSchema, KuduPredicate.ComparisonOp.valueOf(comparisonOp), value);
+ scannerBuilder.addPredicate(predicate);
+ }
+
+ protected void scan(ProcessContext context, ProcessSession session, KuduTable kuduTable, String predicates, List<String> projectedColumnNames, ResultHandler handler) throws Exception {
+ KuduScanner.KuduScannerBuilder scannerBuilder = this.kuduClient.newScannerBuilder(kuduTable);
+ final String[] arrayPredicates = (predicates == null || predicates.isEmpty() ? new String[0] : predicates.split(","));
+
+ for(String column : arrayPredicates){
+ if (column.contains("=")) {
+ final String[] parts = column.split("=");
+ addPredicate(scannerBuilder, kuduTable, parts[0], parts[1], "EQUAL");
+ } else if(column.contains(">")) {
+ final String[] parts = column.split(">");
+ addPredicate(scannerBuilder, kuduTable, parts[0], parts[1], "GREATER");
+ } else if(column.contains("<")) {
+ final String[] parts = column.split("<");
+ addPredicate(scannerBuilder, kuduTable, parts[0], parts[1], "LESS");
+ } else if(column.contains(">=")) {
+ final String[] parts = column.split(">=");
+ addPredicate(scannerBuilder, kuduTable, parts[0], parts[1], "GREATER_EQUAL");
+ } else if(column.contains("<=")) {
+ final String[] parts = column.split("<=");
+ addPredicate(scannerBuilder, kuduTable, parts[0], parts[1], "LESS_EQUAL");
+ }
+ }
+
+ if(!projectedColumnNames.isEmpty()){
+ scannerBuilder.setProjectedColumnNames(projectedColumnNames);
+ }
+
+ KuduScanner scanner = scannerBuilder.build();
+ for (RowResult rowResult: scanner) {
+ handler.handle(rowResult);
+ }
+ }
+
+ /**
+ * Serializes a row from Kudu to a JSON document of the form:
+ *
+ * {
+ * "rows": [
+ * {
+ * "columnname-1" : "value1",
+ * "columnname-2" : "value2",
+ * "columnname-3" : "value3",
+ * "columnname-4" : "value4",
+ * },
+ * {
+ * "columnname-1" : "value1",
+ * "columnname-2" : "value2",
+ * "columnname-3" : "value3",
+ * "columnname-4" : "value4",
+ * }
+ * ]
+ * }
+ */
+ protected String convertToJson(RowResult row) {
Review comment:
Not sure if JSON should be the format of choice here.
Avro seems to be a better option. If it needs to be displayed or processed as a JSON a ConvertRecord processor could help with that.
Also (even if it builds an Avro) it probably should not be in the `AbstractKuduProcessor` but int the `ScanKuduResultHandler` itself.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services