You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by MikeThomsen <gi...@git.apache.org> on 2017/11/24 15:36:47 UTC
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
GitHub user MikeThomsen opened a pull request:
https://github.com/apache/nifi/pull/2294
NIFI-3538 Added DeleteHBaseRow
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [ ] Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [ ] Has your PR been rebased against the latest commit within the target branch (typically master)?
- [ ] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
- [ ] Have you written or updated unit tests to verify your changes?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
- [ ] If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
- [ ] If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/MikeThomsen/nifi NIFI-3538
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/2294.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2294
----
commit 6df9cc009be3ea681d3029e5bc40118a9dd26dd7
Author: Mike Thomsen <mi...@gmail.com>
Date: 2017-11-24T11:06:41Z
NIFI-3538 Added DeleteHBaseRow
----
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167144686
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
+ protected static final PropertyDescriptor HBASE_CLIENT_SERVICE = new PropertyDescriptor.Builder()
+ .name("HBase Client Service")
+ .description("Specifies the Controller Service to use for accessing HBase.")
+ .required(true)
+ .identifiesControllerService(HBaseClientService.class)
+ .build();
+ protected static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the HBase Table to put data into")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ protected static final PropertyDescriptor ROW_ID = new PropertyDescriptor.Builder()
+ .name("Row Identifier")
+ .description("Specifies the Row ID to use when inserting data into HBase")
--- End diff --
Should be "deleting data from HBase".
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157256636
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
--- End diff --
nit: the attribute involved is only one, thus, I'd rename to `doDeleteFromAttribute`
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157383902
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
I'll have to think about the body scenario, but I don't see a good way to do otherwise if the user batches on multiple flowfiles short of hitting HBase with a ton of one off delete statements. We can do that, but my understanding is that HBase really doesn't like write operations to be handled like that.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167152061
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/test/java/org/apache/nifi/hbase/MockHBaseClientService.java ---
@@ -71,6 +73,43 @@ public void delete(String tableName, byte[] rowId) throws IOException {
throw new UnsupportedOperationException();
}
+ @Override
+ public void delete(String tableName, List<byte[]> rowIds) throws IOException {
+ if (throwException) {
+ throw new RuntimeException("Simulated connectivity error");
+ }
+
+ Random random = new Random();
+ int location = 0;
+ if (rowIds.size() > 1) {
+ while (location == 0) {
+ location = random.nextInt(rowIds.size());
--- End diff --
Using random to simulate exception situation looks cool, but it should be avoided in unit testing where we want consistent behavior at every time. I prefer making it a parameter passed from the test method.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167943515
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final String RESTART_INDEX = "restart.index";
+ static final String ROWKEY_START = "rowkey.start";
+ static final String ROWKEY_END = "rowkey.end";
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-batch-size")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("delete-char-set")
+ .displayName("Character Set")
+ .description("The character set used to encode the row key for HBase.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+ properties.add(CHARSET);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ final String charset = context.getProperty(CHARSET).getValue();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ flowFile = doDeleteFromBody(flowFile, context, session, tableName, batchSize, charset);
+ if (flowFile.getAttribute(RESTART_INDEX) != null) {
+ session.transfer(flowFile, REL_FAILURE);
+ } else {
+ final String transitUrl = clientService.toTransitUri(tableName, flowFile.getAttribute(ROWKEY_END));
+ session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().invokeRemoteProcess(flowFile, transitUrl);
+ }
+ } else {
+ String transitUrl = doDeleteFromAttribute(flowFile, context, tableName, charset);
+ session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().invokeRemoteProcess(flowFile, transitUrl);
+ }
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private String doDeleteFromAttribute(FlowFile flowFile, ProcessContext context, String tableName, String charset) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes(charset));
+
+ return clientService.toTransitUri(tableName, rowKey);
+ }
+
+ private FlowFile doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, String tableName, int batchSize, String charset) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).evaluateAttributeExpressions(flowFile).getValue();
+ final String restartIndex = flowFile.getAttribute(RESTART_INDEX);
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ session.exportTo(flowFile, out);
+ out.close();
+
+ String data = new String(out.toByteArray(), charset);
+
+ int restartFrom = -1;
+ if (restartIndex != null) {
+ restartFrom = Integer.parseInt(restartIndex);
+ }
+
+ String first = null, last = null;
+
+ List<byte[]> batch = new ArrayList<>();
+ if (data != null && data.length() > 0) {
+ String[] parts = data.split(keySeparator);
--- End diff --
I updated the documentation because the code is the way to go here since it should be a configurable separator.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167144332
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
+ protected static final PropertyDescriptor HBASE_CLIENT_SERVICE = new PropertyDescriptor.Builder()
+ .name("HBase Client Service")
+ .description("Specifies the Controller Service to use for accessing HBase.")
+ .required(true)
+ .identifiesControllerService(HBaseClientService.class)
+ .build();
+ protected static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the HBase Table to put data into")
--- End diff --
This should be 'to delete'.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167149587
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The character set used to encode the row key for HBase.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+ properties.add(CHARSET);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ final String charset = context.getProperty(CHARSET).getValue();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ flowFile = doDeleteFromBody(flowFile, context, session, batchSize, charset);
+ Relationship rel = flowFile.getAttribute("restart.index") != null ? REL_FAILURE : REL_SUCCESS;
+ session.transfer(flowFile, rel);
+ } else {
+ doDeleteFromAttribute(flowFile, context, charset);
+ session.transfer(flowFile, REL_SUCCESS);
+ }
--- End diff --
It's possible that I overlook but do we send provenance events when this processor deletes rows from HBase? If not, please add that around here. Provenance events `transit url` is important for things like NiFi Atlas integration. [AbstractPutHBase](https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractPutHBase.java#L198) can be a reference, but since it's delete objects, call [ProvenanceReporter.invokeRemoteProcess](https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java#L238) instead.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167858912
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
--- End diff --
I'd suggest keep the naming consistent. Both of `body` and `content` are used, but it would be cleaner if we use just one of them. If so, I'd prefer `content`.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167940501
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167940422
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
--- End diff --
Done.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167148328
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
--- End diff --
What does this "ff" stand for? I wonder if "delete-hb-batch-size" sounds better.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167210043
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
+ protected static final PropertyDescriptor HBASE_CLIENT_SERVICE = new PropertyDescriptor.Builder()
+ .name("HBase Client Service")
+ .description("Specifies the Controller Service to use for accessing HBase.")
+ .required(true)
+ .identifiesControllerService(HBaseClientService.class)
+ .build();
+ protected static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the HBase Table to put data into")
--- End diff --
Done
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
Updated according to the feedback and all tests pass now.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167151202
--- Diff: nifi-nar-bundles/nifi-standard-services/nifi-hbase-client-service-api/src/main/java/org/apache/nifi/hbase/HBaseClientService.java ---
@@ -116,6 +117,13 @@
*/
void delete(String tableName, byte[] rowId) throws IOException;
+ /**
+ * Deletes a list of rows in HBase. All cells are deleted.
+ *
+ */
+
--- End diff --
This blank line can be removed. Also, it might be obvious but I'd expect the same level of documentation with the other methods for Parameters and Exception.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167851490
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/test/java/org/apache/nifi/hbase/TestDeleteHBaseRow.java ---
@@ -0,0 +1,197 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class TestDeleteHBaseRow {
+ private TestRunner runner;
+ private MockHBaseClientService hBaseClient;
+
+ @Before
+ public void setup() throws InitializationException {
+ runner = TestRunners.newTestRunner(new DeleteHBaseRow());
+
+ hBaseClient = new MockHBaseClientService();
+ runner.addControllerService("hbaseClient", hBaseClient);
+ runner.enableControllerService(hBaseClient);
+
+ runner.setProperty(DeleteHBaseRow.TABLE_NAME, "nifi");
+ runner.setProperty(DeleteHBaseRow.HBASE_CLIENT_SERVICE, "hbaseClient");
+ }
+
+ List<String> populateTable(int max) {
+ List<String> ids = new ArrayList<>();
+ for (int index = 0; index < max; index++) {
+ String uuid = UUID.randomUUID().toString();
+ ids.add(uuid);
+ Map<String, String> cells = new HashMap<>();
+ cells.put("test", UUID.randomUUID().toString());
+ hBaseClient.addResult(uuid, cells, System.currentTimeMillis());
+ }
+
+ return ids;
+ }
+
+ @Test
+ public void testSimpleDelete() {
+ List<String> ids = populateTable(100);
+
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "100");
+ runner.setProperty(DeleteHBaseRow.FLOWFILE_FETCH_COUNT, "100");
+ for (String id : ids) {
+ runner.enqueue(id);
+ }
+
+ runner.run(1, true);
+ Assert.assertTrue("The mock client was not empty.", hBaseClient.isEmpty());
+ }
+
+ private String buildSeparatedString(List<String> ids, String separator) {
+ StringBuilder sb = new StringBuilder();
+ for (int index = 1; index <= ids.size(); index++) {
+ sb.append(ids.get(index - 1)).append(separator);
+ }
+
+ return sb.toString();
+ }
+
+ private void testSeparatedDeletes(String separator) {
+ testSeparatedDeletes(separator, separator, new HashMap());
+ }
+
+ private void testSeparatedDeletes(String separator, String separatorProp, Map attrs) {
+ List<String> ids = populateTable(10000);
+ runner.setProperty(DeleteHBaseRow.KEY_SEPARATOR, separator);
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "100");
+ runner.enqueue(buildSeparatedString(ids, separatorProp), attrs);
+ runner.run(1, true);
+
+ Assert.assertTrue("The mock client was not empty.", hBaseClient.isEmpty());
+ }
+
+ @Test
+ public void testDeletesSeparatedByNewLines() {
+ testSeparatedDeletes("\n");
+ }
+
+ @Test
+ public void testDeletesSeparatedByCommas() {
+ testSeparatedDeletes(",");
+ }
+
+ @Test
+ public void testDeleteWithELSeparator() {
+ runner.setValidateExpressionUsage(true);
+ Map<String, String> attrs = new HashMap<>();
+ attrs.put("test.separator", "____");
+ testSeparatedDeletes("${test.separator}", "____", attrs);
+ }
+
+ @Test
+ public void testDeleteWithExpressionLanguage() {
+ List<String> ids = populateTable(1000);
+ for (String id : ids) {
+ String[] parts = id.split("-");
+ Map<String, String> attrs = new HashMap<>();
+ for (int index = 0; index < parts.length; index++) {
+ attrs.put(String.format("part_%d", index), parts[index]);
+ }
+ runner.enqueue(id, attrs);
+ }
+ runner.setProperty(DeleteHBaseRow.ROW_ID, "${part_0}-${part_1}-${part_2}-${part_3}-${part_4}");
+ runner.setProperty(DeleteHBaseRow.ROW_ID_LOCATION, DeleteHBaseRow.ROW_ID_ATTR);
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "200");
+ runner.setValidateExpressionUsage(true);
+ runner.run(1, true);
+ }
+
+ @Test
+ public void testConnectivityErrorHandling() {
+ List<String> ids = populateTable(100);
+ for (String id : ids) {
+ runner.enqueue(id);
+ }
+ boolean exception = false;
+ try {
+ hBaseClient.setThrowException(true);
+ runner.run(1, true);
+ } catch (Exception ex) {
+ exception = true;
+ } finally {
+ hBaseClient.setThrowException(false);
+ }
+
+ Assert.assertFalse("An unhandled exception was caught.", exception);
+ }
+
+ @Test
+ public void testRestartIndexAttribute() {
+ List<String> ids = populateTable(500);
+ StringBuilder sb = new StringBuilder();
+ for (int index = 0; index < ids.size(); index++) {
+ sb.append(ids.get(index)).append( index < ids.size() - 1 ? "," : "");
+ }
+ runner.enqueue(sb.toString());
+ runner.setProperty(DeleteHBaseRow.ROW_ID_LOCATION, DeleteHBaseRow.ROW_ID_BODY);
+
+ Assert.assertTrue("There should have been 500 rows.", hBaseClient.size() == 500);
+
+ hBaseClient.setDeletePoint(20);
+ hBaseClient.setThrowExceptionDuringBatchDelete(true);
+ runner.run(1, true, true);
+
+ runner.assertTransferCount(DeleteHBaseRow.REL_FAILURE, 1);
+ runner.assertTransferCount(DeleteHBaseRow.REL_SUCCESS, 0);
+
+ Assert.assertTrue("Nothing was deleted", hBaseClient.size() < 500);
--- End diff --
The assertion message should be "Partially deleted".
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157368551
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
--- End diff --
It might be, but I've never been told to change it before. If someone wants to make a real issue of it, I'll change it. Otherwise, I think it helps with readability in cases like this.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r160140592
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,154 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFile, context, session, batchSize);
+ } else {
+ doDeleteFromAttribute(flowFile, context);
+ }
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttribute(FlowFile flowFile, ProcessContext context) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes("UTF-8"));
+ }
+
+ private void doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+
+ List<byte[]> deletes = new ArrayList<>();
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ if (data != null && data.length() > 0) {
+ String[] parts = data.split(keySeparator);
+ for (int index = 0; index < parts.length; index++) {
+ deletes.add(parts[index].getBytes("UTF-8"));
+ }
+ }
+ });
+
+
+ List<byte[]> batch = new ArrayList<>();
+ for (int index = 0; index < deletes.size(); index++) {
--- End diff --
this for loop can be avoided and it can be merged in the previous one, can't it?
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on the issue:
https://github.com/apache/nifi/pull/2294
LGTM
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
That restart.index change is in place. I put a unit test in there that tests it out and it works just fine.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167941736
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167854208
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
--- End diff --
It says "Only written when a flowfile contains multiple deletes." but the attribute is written when a FlowFile only contains single row id. It would be more accurate if we say "Only written when deleting Row IDs from flowfile content."
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167319106
--- Diff: nifi-nar-bundles/nifi-standard-services/nifi-hbase-client-service-api/src/main/java/org/apache/nifi/hbase/HBaseClientService.java ---
@@ -116,6 +117,13 @@
*/
void delete(String tableName, byte[] rowId) throws IOException;
+ /**
+ * Deletes a list of rows in HBase. All cells are deleted.
+ *
+ */
+
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r160178312
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,154 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFile, context, session, batchSize);
+ } else {
+ doDeleteFromAttribute(flowFile, context);
+ }
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttribute(FlowFile flowFile, ProcessContext context) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes("UTF-8"));
+ }
+
+ private void doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+
+ List<byte[]> deletes = new ArrayList<>();
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ if (data != null && data.length() > 0) {
+ String[] parts = data.split(keySeparator);
+ for (int index = 0; index < parts.length; index++) {
+ deletes.add(parts[index].getBytes("UTF-8"));
+ }
+ }
+ });
+
+
+ List<byte[]> batch = new ArrayList<>();
+ for (int index = 0; index < deletes.size(); index++) {
--- End diff --
I'll take a closer look.
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by joewitt <gi...@git.apache.org>.
Github user joewitt commented on the issue:
https://github.com/apache/nifi/pull/2294
Looks like there is some review feedback and Travis-CI fails due to checkstyle issues. Can you please address?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r159047906
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,158 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFile, context, session, batchSize);
+ } else {
+ doDeleteFromAttribute(flowFile, context);
+ }
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttribute(FlowFile flowFile, ProcessContext context) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes("UTF-8"));
+ }
+
+ private void doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+
+ List<byte[]> deletes = new ArrayList<>();
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
--- End diff --
we already have the separator, do we need this? Moreover if the line terminator is `\r\n` this would generate an empty key to be deleted too IIUC. Since HBase keys are bytes `\r` and `\n` may be valid values inside a key the user wants to delete... I'd avoid doing this.
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@ijokarumawak All of your feedback should now be addressed.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157258216
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
I think this is quite a dangerous behavior. This might lead to situation in which part of the flowfiles processed are succeeded and some not. Can't we just submit one batch per run? The user can control the batch size using FLOWFILE_FETCH_COUNT...
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on the issue:
https://github.com/apache/nifi/pull/2294
@MikeThomsen there are validation errors due to unused imports. May you please fix them?
```
[[1;33mWARNING[m] src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java:[31,8] (imports) UnusedImports: Unused import - java.io.IOException.
[[1;33mWARNING[m] src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java:[33,8] (imports) UnusedImports: Unused import - java.util.HashMap.
[[1;33mWARNING[m] src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java:[35,8] (imports) UnusedImports: Unused import - java.util.Map.
```
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167210016
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
+ protected static final PropertyDescriptor HBASE_CLIENT_SERVICE = new PropertyDescriptor.Builder()
+ .name("HBase Client Service")
+ .description("Specifies the Controller Service to use for accessing HBase.")
+ .required(true)
+ .identifiesControllerService(HBaseClientService.class)
+ .build();
+ protected static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the HBase Table to put data into")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ protected static final PropertyDescriptor ROW_ID = new PropertyDescriptor.Builder()
+ .name("Row Identifier")
+ .description("Specifies the Row ID to use when inserting data into HBase")
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167851239
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
--- End diff --
`@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)` should be added.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167857406
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final String RESTART_INDEX = "restart.index";
+ static final String ROWKEY_START = "rowkey.start";
+ static final String ROWKEY_END = "rowkey.end";
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-batch-size")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("delete-char-set")
+ .displayName("Character Set")
+ .description("The character set used to encode the row key for HBase.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+ properties.add(CHARSET);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ final String charset = context.getProperty(CHARSET).getValue();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ flowFile = doDeleteFromBody(flowFile, context, session, tableName, batchSize, charset);
+ if (flowFile.getAttribute(RESTART_INDEX) != null) {
+ session.transfer(flowFile, REL_FAILURE);
+ } else {
+ final String transitUrl = clientService.toTransitUri(tableName, flowFile.getAttribute(ROWKEY_END));
+ session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().invokeRemoteProcess(flowFile, transitUrl);
+ }
+ } else {
+ String transitUrl = doDeleteFromAttribute(flowFile, context, tableName, charset);
+ session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().invokeRemoteProcess(flowFile, transitUrl);
+ }
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private String doDeleteFromAttribute(FlowFile flowFile, ProcessContext context, String tableName, String charset) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes(charset));
+
+ return clientService.toTransitUri(tableName, rowKey);
+ }
+
+ private FlowFile doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, String tableName, int batchSize, String charset) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).evaluateAttributeExpressions(flowFile).getValue();
+ final String restartIndex = flowFile.getAttribute(RESTART_INDEX);
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ session.exportTo(flowFile, out);
+ out.close();
+
+ String data = new String(out.toByteArray(), charset);
+
+ int restartFrom = -1;
+ if (restartIndex != null) {
+ restartFrom = Integer.parseInt(restartIndex);
+ }
+
+ String first = null, last = null;
+
+ List<byte[]> batch = new ArrayList<>();
+ if (data != null && data.length() > 0) {
+ String[] parts = data.split(keySeparator);
--- End diff --
The usage says "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, row IDs separated by commas or a combination of the two."
However, the code does not treat line separator. Should we update the doc or code?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157414765
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
yes, I think there isn't a perfect solution. Though, if the user sees that there are performance problems using a lot of small flowfiles, he can always put the keys in the body and merge the flowfiles, thus obtaining better performances. I think that consistency and correctness of the outcome are more important than performance. What do you think?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157255826
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
--- End diff --
what about using `defaultValue(ROW_ID_BODY.getValue())`?
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@ijokarumawak @bbende The user who posted on nifi-dev asking for a copy of this confirmed that he's using it in production and is happy with it. If you have any time, could you take a quick look and see if you think it's ready for merge?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167150836
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The character set used to encode the row key for HBase.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+ properties.add(CHARSET);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ final String charset = context.getProperty(CHARSET).getValue();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ flowFile = doDeleteFromBody(flowFile, context, session, batchSize, charset);
+ Relationship rel = flowFile.getAttribute("restart.index") != null ? REL_FAILURE : REL_SUCCESS;
+ session.transfer(flowFile, rel);
+ } else {
+ doDeleteFromAttribute(flowFile, context, charset);
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
--- End diff --
Wouldn't it be useful if we add error message and exception type to the FlowFile attribute so that user can define more conditional retry flow?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167144543
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
--- End diff --
Is there any reason you want this abstract layer? Since it does not have variations like PutHBase does, I'd prefer move these code to DeleteHBaseRow for now, we don't know which should be abstracted, yet.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r159047973
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,158 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFile, context, session, batchSize);
+ } else {
+ doDeleteFromAttribute(flowFile, context);
+ }
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception ex) {
+ getLogger().error(ex.getMessage(), ex);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttribute(FlowFile flowFile, ProcessContext context) throws Exception {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ clientService.delete(tableName, rowKey.getBytes("UTF-8"));
+ }
+
+ private void doDeleteFromBody(FlowFile flowFile, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ String keySeparator = context.getProperty(KEY_SEPARATOR).getValue();
--- End diff --
can we add EL support to this field?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167148081
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
--- End diff --
Please use `name` and `displayName` to align with other properties.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167209979
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167939817
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
+ }
+)
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167854611
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,213 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttributes(
+ value = {
+ @WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' " +
+ "relationship, so that this processor can retry from there when the same FlowFile is routed again." ),
+ @WritesAttribute( attribute = "rowkey.start", description = "The first rowkey in the flowfile. Only written when a flowfile contains multiple deletes."),
+ @WritesAttribute( attribute = "rowkey.end", description = "The last rowkey in the flowfile. Only written when a flowfile contains multiple deletes.")
--- End diff --
Same comment as "rowkey.start" on the description applied here.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157383920
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
**batches on multiple flowfiles using attributes instead of the body.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157257248
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
--- End diff --
what if the user wants to delete a row which contains a comma?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167195155
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
--- End diff --
I was planning to do a DeleteHBaseRecords processor later.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167940095
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/test/java/org/apache/nifi/hbase/TestDeleteHBaseRow.java ---
@@ -0,0 +1,197 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class TestDeleteHBaseRow {
+ private TestRunner runner;
+ private MockHBaseClientService hBaseClient;
+
+ @Before
+ public void setup() throws InitializationException {
+ runner = TestRunners.newTestRunner(new DeleteHBaseRow());
+
+ hBaseClient = new MockHBaseClientService();
+ runner.addControllerService("hbaseClient", hBaseClient);
+ runner.enableControllerService(hBaseClient);
+
+ runner.setProperty(DeleteHBaseRow.TABLE_NAME, "nifi");
+ runner.setProperty(DeleteHBaseRow.HBASE_CLIENT_SERVICE, "hbaseClient");
+ }
+
+ List<String> populateTable(int max) {
+ List<String> ids = new ArrayList<>();
+ for (int index = 0; index < max; index++) {
+ String uuid = UUID.randomUUID().toString();
+ ids.add(uuid);
+ Map<String, String> cells = new HashMap<>();
+ cells.put("test", UUID.randomUUID().toString());
+ hBaseClient.addResult(uuid, cells, System.currentTimeMillis());
+ }
+
+ return ids;
+ }
+
+ @Test
+ public void testSimpleDelete() {
+ List<String> ids = populateTable(100);
+
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "100");
+ runner.setProperty(DeleteHBaseRow.FLOWFILE_FETCH_COUNT, "100");
+ for (String id : ids) {
+ runner.enqueue(id);
+ }
+
+ runner.run(1, true);
+ Assert.assertTrue("The mock client was not empty.", hBaseClient.isEmpty());
+ }
+
+ private String buildSeparatedString(List<String> ids, String separator) {
+ StringBuilder sb = new StringBuilder();
+ for (int index = 1; index <= ids.size(); index++) {
+ sb.append(ids.get(index - 1)).append(separator);
+ }
+
+ return sb.toString();
+ }
+
+ private void testSeparatedDeletes(String separator) {
+ testSeparatedDeletes(separator, separator, new HashMap());
+ }
+
+ private void testSeparatedDeletes(String separator, String separatorProp, Map attrs) {
+ List<String> ids = populateTable(10000);
+ runner.setProperty(DeleteHBaseRow.KEY_SEPARATOR, separator);
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "100");
+ runner.enqueue(buildSeparatedString(ids, separatorProp), attrs);
+ runner.run(1, true);
+
+ Assert.assertTrue("The mock client was not empty.", hBaseClient.isEmpty());
+ }
+
+ @Test
+ public void testDeletesSeparatedByNewLines() {
+ testSeparatedDeletes("\n");
+ }
+
+ @Test
+ public void testDeletesSeparatedByCommas() {
+ testSeparatedDeletes(",");
+ }
+
+ @Test
+ public void testDeleteWithELSeparator() {
+ runner.setValidateExpressionUsage(true);
+ Map<String, String> attrs = new HashMap<>();
+ attrs.put("test.separator", "____");
+ testSeparatedDeletes("${test.separator}", "____", attrs);
+ }
+
+ @Test
+ public void testDeleteWithExpressionLanguage() {
+ List<String> ids = populateTable(1000);
+ for (String id : ids) {
+ String[] parts = id.split("-");
+ Map<String, String> attrs = new HashMap<>();
+ for (int index = 0; index < parts.length; index++) {
+ attrs.put(String.format("part_%d", index), parts[index]);
+ }
+ runner.enqueue(id, attrs);
+ }
+ runner.setProperty(DeleteHBaseRow.ROW_ID, "${part_0}-${part_1}-${part_2}-${part_3}-${part_4}");
+ runner.setProperty(DeleteHBaseRow.ROW_ID_LOCATION, DeleteHBaseRow.ROW_ID_ATTR);
+ runner.setProperty(DeleteHBaseRow.BATCH_SIZE, "200");
+ runner.setValidateExpressionUsage(true);
+ runner.run(1, true);
+ }
+
+ @Test
+ public void testConnectivityErrorHandling() {
+ List<String> ids = populateTable(100);
+ for (String id : ids) {
+ runner.enqueue(id);
+ }
+ boolean exception = false;
+ try {
+ hBaseClient.setThrowException(true);
+ runner.run(1, true);
+ } catch (Exception ex) {
+ exception = true;
+ } finally {
+ hBaseClient.setThrowException(false);
+ }
+
+ Assert.assertFalse("An unhandled exception was caught.", exception);
+ }
+
+ @Test
+ public void testRestartIndexAttribute() {
+ List<String> ids = populateTable(500);
+ StringBuilder sb = new StringBuilder();
+ for (int index = 0; index < ids.size(); index++) {
+ sb.append(ids.get(index)).append( index < ids.size() - 1 ? "," : "");
+ }
+ runner.enqueue(sb.toString());
+ runner.setProperty(DeleteHBaseRow.ROW_ID_LOCATION, DeleteHBaseRow.ROW_ID_BODY);
+
+ Assert.assertTrue("There should have been 500 rows.", hBaseClient.size() == 500);
+
+ hBaseClient.setDeletePoint(20);
+ hBaseClient.setThrowExceptionDuringBatchDelete(true);
+ runner.run(1, true, true);
+
+ runner.assertTransferCount(DeleteHBaseRow.REL_FAILURE, 1);
+ runner.assertTransferCount(DeleteHBaseRow.REL_SUCCESS, 0);
+
+ Assert.assertTrue("Nothing was deleted", hBaseClient.size() < 500);
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167147850
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/AbstractDeleteHBase.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public abstract class AbstractDeleteHBase extends AbstractProcessor {
+ protected static final PropertyDescriptor HBASE_CLIENT_SERVICE = new PropertyDescriptor.Builder()
+ .name("HBase Client Service")
+ .description("Specifies the Controller Service to use for accessing HBase.")
+ .required(true)
+ .identifiesControllerService(HBaseClientService.class)
+ .build();
+ protected static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the HBase Table to put data into")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ protected static final PropertyDescriptor ROW_ID = new PropertyDescriptor.Builder()
+ .name("Row Identifier")
+ .description("Specifies the Row ID to use when inserting data into HBase")
+ .required(false) // not all sub-classes will require this
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
--- End diff --
It would be more user friendly if we add custom validate to suggest setting this property when `Row Id Location` is `FlowFIle attributes`
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/2294
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@mgaido91 I added the charset. I can't believe I missed that...
WRT the batching, I stand by my opinion that we need to have a sane default there because there needs to be a way to ensure someone doesn't accidentally (or on purpose) send an operation that is too big to HBase at once.
To me this is not a theoretical issue because I ran into something like this with PutHBaseRecord doing genomic data ingestion w/ NiFi. The data set would generate easily 10B, if not 20-25B tiny (like few dozen byte) writes. I had to really scale back the size of each record set I was sending to PutHBaseRecord because it was easily to generate so many Puts that it would hammer a region offline unexpectedly.
I'm not a HBase expert by any means, but it seems like a recipe for trouble based on my experience with putting a lot of small writes (and Delete objects are tiny writes).
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@mgaido91 That loop, incidentally, is being eliminated now because I decided to follow the same practice that was used in PutHBaseRecord where on failure it adds "restart.index" which is roughly where in the loop it failed.
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on the issue:
https://github.com/apache/nifi/pull/2294
thanks @MikeThomsen. Actually I was referring at [this comment](https://github.com/apache/nifi/pull/2294#discussion_r160140592). I think that is my last comment. Overall it LGTM.
On the batch size, I think that the committers who will have to review this before merging can express their opinion. Your opinion makes sense, but it involves a consistency issue: we need a tradeoff and I think only committers can decide what to do.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157368648
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
--- End diff --
Since this is a delete processor, you may be right. This would make more sense in a put because the scenarios where a user is likely to do all spaces by accident are more likely than intentional use.
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@mgaido91
I'll make the UTF-8 charset a configuration parameter.
WRT the batch issue, my preference for providing a batch option with a sane default size comes from seeing less than optimal things involving how much work is sent in a single batch. Trying to be polite on that point, but I think you can get an idea of what I mean...
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167150445
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
--- End diff --
This description can be improved. It's possible that user might expect restart happens automatically. How about something like:
"If a delete batch fails, 'restart.index' attribute is added to the FlowFile and sent to 'failure' relationship, so that this processor can retry from there when the same FlowFile is routed again."
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167307609
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor KEY_SEPARATOR = new PropertyDescriptor.Builder()
+ .name("delete-hb-separator")
+ .displayName("Delete Row Key Separator")
+ .description("The separator character(s) that separate multiple row keys " +
+ "when multiple row keys are provided in the flowfile body")
+ .required(true)
+ .defaultValue(",")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The character set used to encode the row key for HBase.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+ properties.add(KEY_SEPARATOR);
+ properties.add(CHARSET);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ final String charset = context.getProperty(CHARSET).getValue();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ for (int index = 0; index < flowFiles.size(); index++) {
+ FlowFile flowFile = flowFiles.get(index);
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ flowFile = doDeleteFromBody(flowFile, context, session, batchSize, charset);
+ Relationship rel = flowFile.getAttribute("restart.index") != null ? REL_FAILURE : REL_SUCCESS;
+ session.transfer(flowFile, rel);
+ } else {
+ doDeleteFromAttribute(flowFile, context, charset);
+ session.transfer(flowFile, REL_SUCCESS);
+ }
--- End diff --
Done
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on the issue:
https://github.com/apache/nifi/pull/2294
@MikeThomsen Thank you for the quick updates! All LGTM now, +1. Merging to master.
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@ijokarumawak I think all of your concerns have been addressed now.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157377689
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
does this mean that we are telling the user that some flowfile failed even though they were successful? I think the best approach here would be to process one flowfile per time: in this way the user can be sure of what happened. What do you think?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167319088
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/test/java/org/apache/nifi/hbase/MockHBaseClientService.java ---
@@ -71,6 +73,43 @@ public void delete(String tableName, byte[] rowId) throws IOException {
throw new UnsupportedOperationException();
}
+ @Override
+ public void delete(String tableName, List<byte[]> rowIds) throws IOException {
+ if (throwException) {
+ throw new RuntimeException("Simulated connectivity error");
+ }
+
+ Random random = new Random();
+ int location = 0;
+ if (rowIds.size() > 1) {
+ while (location == 0) {
+ location = random.nextInt(rowIds.size());
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157256381
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
--- End diff --
I think that the preferred syntax in the project is to have only one space before and after the `=` sign
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on the issue:
https://github.com/apache/nifi/pull/2294
@bbende @ijokarumawak If either of you get a chance, could you take a look at this? I'm working on a big change to the HBase capabilities that would add visibility label support and would like to be able to deliver that w/ delete functionality built in to NiFi users :)
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157256777
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
--- End diff --
why can't a key be made of spaces?
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167209940
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
--- End diff --
Done
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157368968
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
If one of the delete fails because a region is not online or something like that, all of the flowfiles will be transferred to the failure relationship.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r157368931
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,178 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue("body")
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
+ .displayName("Batch Size")
+ .description("The number of deletes to send per batch.")
+ .required(true)
+ .defaultValue("50")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = super.getSupportedPropertyDescriptors();
+ properties.add(ROW_ID_LOCATION);
+ properties.add(FLOWFILE_FETCH_COUNT);
+ properties.add(BATCH_SIZE);
+
+ return properties;
+ }
+
+ @Override
+ protected void doDelete(ProcessContext context, ProcessSession session) throws Exception {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final String location = context.getProperty(ROW_ID_LOCATION).getValue();
+ final int flowFileCount = context.getProperty(FLOWFILE_FETCH_COUNT).asInteger();
+ List<FlowFile> flowFiles = session.get(flowFileCount);
+
+ if (flowFiles != null && flowFiles.size() > 0) {
+ try {
+ if (location.equals(ROW_ID_BODY.getValue())) {
+ doDeleteFromBody(flowFiles, context, session, batchSize);
+ } else {
+ doDeleteFromAttributes(flowFiles, context, session, batchSize);
+ }
+
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_SUCCESS);
+ }
+ } catch (Exception ex) {
+ getLogger().error("");
+ for (int index = 0; index < flowFiles.size(); index++) {
+ session.transfer(flowFiles.get(index), REL_FAILURE);
+ }
+ }
+ }
+ }
+
+ private void doDeleteFromAttributes(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ String rowKey = context.getProperty(ROW_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ if (rowKey == null || rowKey.trim().equals("")) {
+ throw new RuntimeException("Could not find a row key with provided EL statement, stopping...");
+ }
+
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+
+ batch.add(rowKey.getBytes("UTF-8"));
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void doDeleteFromBody(List<FlowFile> flowFiles, ProcessContext context, ProcessSession session, int batchSize) throws Exception {
+ Map<String, List<byte[]>> deletes = new HashMap<>();
+ for (FlowFile flowFile : flowFiles) {
+ session.read(flowFile, in -> {
+ String data = IOUtils.toString(in, "UTF-8");
+ String[] lines = data.split("[\\r\\n]");
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ for (String line : lines) {
+ String[] separated = line.split(",");
+ for (String part : separated) {
+ List<byte[]> batch = deletes.get(tableName);
+ if (batch == null) {
+ batch = new ArrayList<>();
+ deletes.put(tableName, batch);
+ }
+ batch.add(part.trim().getBytes("UTF-8"));
+ }
+ }
+ });
+ }
+
+ processDeletes(deletes, batchSize);
+ }
+
+ private void processDeletes(Map<String, List<byte[]>> deletes, int batchSize) throws IOException {
+ for (Map.Entry<String, List<byte[]>> entry : deletes.entrySet()) {
+ if (entry.getValue().size() <= batchSize) {
+ clientService.delete(entry.getKey(), entry.getValue());
+ } else {
--- End diff --
This follows the pattern of other HBase processors in allowing the table to be configured with EL. So we can't do one big batch since it's possible the user will a group of flowfiles that will go against multiple tables.
---
[GitHub] nifi pull request #2294: NIFI-3538 Added DeleteHBaseRow
Posted by MikeThomsen <gi...@git.apache.org>.
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2294#discussion_r167209961
--- Diff: nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/DeleteHBaseRow.java ---
@@ -0,0 +1,183 @@
+/*
+ * 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.nifi.hbase;
+
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+@WritesAttribute( attribute = "restart.index", description = "If a delete batch fails, it will restart from restart.index" )
+@Tags({ "delete", "hbase" })
+@CapabilityDescription(
+ "Delete HBase records individually or in batches. The input can be a single row ID in the body, one ID per line, " +
+ "row IDs separated by commas or a combination of the two. ")
+public class DeleteHBaseRow extends AbstractDeleteHBase {
+ static final AllowableValue ROW_ID_BODY = new AllowableValue("body", "FlowFile content", "Get the row key(s) from the flowfile content.");
+ static final AllowableValue ROW_ID_ATTR = new AllowableValue("attr", "FlowFile attributes", "Get the row key from an expression language statement.");
+
+ static final PropertyDescriptor ROW_ID_LOCATION = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-id-location")
+ .displayName("Row ID Location")
+ .description("The location of the row ID to use for building the delete. Can be from the content or an expression language statement.")
+ .required(true)
+ .defaultValue(ROW_ID_BODY.getValue())
+ .allowableValues(ROW_ID_BODY, ROW_ID_ATTR)
+ .addValidator(Validator.VALID)
+ .build();
+
+ static final PropertyDescriptor FLOWFILE_FETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("delete-hb-flowfile-fetch-count")
+ .displayName("Flowfile Fetch Count")
+ .description("The number of flowfiles to fetch per run.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .expressionLanguageSupported(false)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("delete-hb-row-ff-count")
--- End diff --
Done
---
[GitHub] nifi issue #2294: NIFI-3538 Added DeleteHBaseRow
Posted by mgaido91 <gi...@git.apache.org>.
Github user mgaido91 commented on the issue:
https://github.com/apache/nifi/pull/2294
@MikeThomsen sorry may you please address the last comment? And add the charset parameter? Thanks.
---