Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

NIFI-6009 ScanKudu Processor #3611

Closed
wants to merge 21 commits into from
Closed

NIFI-6009 ScanKudu Processor #3611

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
e702eab
Kudu Put Operations
SandishKumarHN Jul 29, 2019
a967964
Kudu Scan Processor
SandishKumarHN Jul 29, 2019
d8a9d1a
KuduScannaer for loop instead of while with iterator
SandishKumarHN Aug 1, 2019
7f09403
Merge remote-tracking branch 'remotes/upstream/master' into NIFI-6009…
SandishKumarHN Sep 2, 2019
d28676f
merging with branch
SandishKumarHN Sep 2, 2019
99a03a0
merge changes
SandishKumarHN Sep 2, 2019
224a15c
merge changes
SandishKumarHN Sep 3, 2019
8de7d6a
merge changes
SandishKumarHN Sep 4, 2019
4638933
review changes 1
SandishKumarHN Sep 6, 2019
c6187ea
review changes - 2
SandishKumarHN Sep 22, 2019
6b6a3b3
review changes - 2
SandishKumarHN Sep 22, 2019
b5b4559
review changes - 3
SandishKumarHN Sep 24, 2019
47f9f6b
using kudu test harness for unittesting
SandishKumarHN Sep 25, 2019
ca732e8
review changes - 4
SandishKumarHN Sep 25, 2019
e04e43f
review changes - 5
SandishKumarHN Sep 27, 2019
c6cf293
review changes - 5
SandishKumarHN Sep 27, 2019
3ed1078
review changes - 6
SandishKumarHN Oct 1, 2019
74f18ba
review changes - 7
SandishKumarHN Oct 2, 2019
133f4d6
review changes - 8
SandishKumarHN Oct 3, 2019
c1c9d04
review changes - 8
SandishKumarHN Oct 3, 2019
203c4a1
review changes
Oct 9, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
82 changes: 81 additions & 1 deletion nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-processors/pom.xml
View file
Open in desktop
Expand Up @@ -23,6 +23,10 @@

<artifactId>nifi-kudu-processors</artifactId>
<packaging>jar</packaging>
<properties>
<exclude.tests>None</exclude.tests>
<kudu.version>1.10.0</kudu.version>
</properties>

<dependencies>
<dependency>
Expand All @@ -46,7 +50,13 @@
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-client</artifactId>
<version>1.10.0</version>
<version>${kudu.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-test-utils</artifactId>
<version>${kudu.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
Expand Down Expand Up @@ -87,4 +97,74 @@
<scope>test</scope>
</dependency>
</dependencies>
<build>
<extensions>
<!-- Used to find the right kudu-binary artifact with the Maven
property ${os.detected.classifier} -->
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.6.2</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<exclude>${exclude.tests}</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>kudu-windows</id>
<activation>
<os>
<family>Windows</family>
</os>
</activation>
<properties>
<!-- Kudu tests do not support Windows. -->
<exclude.tests>**/*.java</exclude.tests>
</properties>
</profile>
<profile>
<id>kudu-linux</id>
<activation>
<os>
<family>Unix</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-binary</artifactId>
<version>${kudu.version}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>test</scope>
</dependency>
</dependencies>
</profile>
<profile>
<id>kudu-mac</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-binary</artifactId>
<version>${kudu.version}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>test</scope>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
94 changes: 93 additions & 1 deletion ...-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/AbstractKuduProcessor.java
View file
Open in desktop
Expand Up @@ -17,6 +17,7 @@

package org.apache.nifi.processors.kudu;

import java.security.PrivilegedExceptionAction;
import org.apache.kudu.shaded.com.google.common.annotations.VisibleForTesting;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
Expand All @@ -29,6 +30,7 @@
import org.apache.kudu.client.OperationResponse;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.RowError;
import org.apache.kudu.client.RowResult;
import org.apache.kudu.client.SessionConfiguration;
import org.apache.kudu.client.Delete;
import org.apache.kudu.client.Insert;
Expand All @@ -41,6 +43,8 @@
import org.apache.nifi.kerberos.KerberosCredentialsService;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.security.krb.KerberosAction;
import org.apache.nifi.security.krb.KerberosKeytabUser;
Expand All @@ -50,6 +54,7 @@
import javax.security.auth.login.LoginException;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.List;

Expand Down Expand Up @@ -119,6 +124,25 @@ public void createKuduClient(ProcessContext context) throws LoginException {
this.kuduClient = kerberosAction.execute();
}

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final KerberosUser user = getKerberosUser();

if (user == null) {
trigger(context, session);
return;
}

final PrivilegedExceptionAction<Void> privelegedAction = () -> {
trigger(context, session);
return null;
};

final KerberosAction<Void> action = new KerberosAction<>(user, privelegedAction, getLogger());
action.execute();
}

public abstract void trigger(ProcessContext context, ProcessSession session);

protected KuduClient buildClient(final String masters, final ProcessContext context) {
final Integer operationTimeout = context.getProperty(KUDU_OPERATION_TIMEOUT_MS).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
Expand Down Expand Up @@ -260,4 +284,72 @@ protected Update updateRecordToKudu(KuduTable kuduTable, Record record, List<Str
return update;
}

}

/**
* Serializes a row from Kudu to a JSON document of the form:
*
* {
* "rows": [
* {
* "columnname-1" : "value1",
* "columnname-2" : "value2",
* "columnname-3" : "value3",
* "columnname-4" : "value4",
* },
* {
* "columnname-1" : "value1",
* "columnname-2" : "value2",
* "columnname-3" : "value3",
* "columnname-4" : "value4",
* }
* ]
* }
*/
protected String convertToJson(RowResult row) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if JSON should be the format of choice here.
Avro seems to be a better option. If it needs to be displayed or processed as a JSON a ConvertRecord processor could help with that.

Also (even if it builds an Avro) it probably should not be in the AbstractKuduProcessor but int the ScanKuduResultHandler itself.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm thinking for initial release JSON would be a better choice, as it gives users better transparency and easy for debugging. if required user can use ConvertJSONToAvro.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One of the issue is that JSON is a very low-performance format. With this implementation even if one wants to transform the data into Avro for example, all the serialization-desrialization need to happen for all JSON documents.

Also the JSON generation is basically duplicated here (in a simple way). Most processors that are generating data build Record objects and delegate its serialization to a RecordWriter controller service.
I think following that pattern would be most preferable in this case as well.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method has some issues that needs to be fixed:

  1. INT8 should be retrieved via getByte
  2. INT16 should be retrieved via getShort
  3. DOUBLE is not handled
  4. BINARY now returns a toString of a ByteBuffer

Here are my suggested fixes, although how BINARY should be handled may be up for debate:

    protected String convertToJson(RowResult row) {
        final StringBuilder jsonBuilder = new StringBuilder();
        jsonBuilder.append("{\"rows\":[{");
        Iterator<ColumnSchema> columns = row.getSchema().getColumns().iterator();
        while (columns.hasNext()) {
            ColumnSchema col = columns.next();
            jsonBuilder.append("\"" + col.getName() + "\":");
            switch (col.getType()) {
                case STRING:
                    jsonBuilder.append("\"" + row.getString(col.getName()) + "\"");
                    break;
                case INT8:
                    jsonBuilder.append("\"" + row.getByte(col.getName()) + "\"");
                    break;
                case INT16:
                    jsonBuilder.append("\"" + row.getShort(col.getName()) + "\"");
                    break;
                case INT32:
                    jsonBuilder.append("\"" + row.getInt(col.getName()) + "\"");
                    break;
                case INT64:
                    jsonBuilder.append("\"" + row.getLong(col.getName()) + "\"");
                    break;
                case BOOL:
                    jsonBuilder.append("\"" + row.getBoolean(col.getName()) + "\"");
                    break;
                case DECIMAL:
                    jsonBuilder.append("\"" + row.getDecimal(col.getName()) + "\"");
                    break;
                case FLOAT:
                    jsonBuilder.append("\"" + row.getFloat(col.getName()) + "\"");
                    break;
                case DOUBLE:
                    jsonBuilder.append("\"" + row.getDouble(col.getName()) + "\"");
                    break;
                case UNIXTIME_MICROS:
                    jsonBuilder.append("\"" + row.getLong(col.getName()) + "\"");
                    break;
                case BINARY:
                    jsonBuilder.append("\"0x" + Hex.encodeHexString(row.getBinaryCopy(col.getName())) + "\"");
                    break;
                default:
                    break;
            }
            if(columns.hasNext())
                jsonBuilder.append(",");
        }
        jsonBuilder.append("}]}");
        return jsonBuilder.toString();
    }

final StringBuilder jsonBuilder = new StringBuilder();
jsonBuilder.append("{\"rows\":[{");
Iterator<ColumnSchema> columns = row.getSchema().getColumns().iterator();
while (columns.hasNext()) {
ColumnSchema col = columns.next();
jsonBuilder.append("\"" + col.getName() + "\":");
switch (col.getType()) {
case STRING:
jsonBuilder.append("\"" + row.getString(col.getName()) + "\"");
break;
case INT8:
jsonBuilder.append("\"" + row.getInt(col.getName()) + "\"");
break;
case INT16:
jsonBuilder.append("\"" + row.getInt(col.getName()) + "\"");
break;
case INT32:
jsonBuilder.append("\"" + row.getInt(col.getName()) + "\"");
break;
case INT64:
jsonBuilder.append("\"" + row.getLong(col.getName()) + "\"");
break;
case BOOL:
jsonBuilder.append("\"" + row.getBoolean(col.getName()) + "\"");
break;
case DECIMAL:
jsonBuilder.append("\"" + row.getDecimal(col.getName()) + "\"");
break;
case FLOAT:
jsonBuilder.append("\"" + row.getFloat(col.getName()) + "\"");
break;
case UNIXTIME_MICROS:
jsonBuilder.append("\"" + row.getLong(col.getName()) + "\"");
break;
case BINARY:
jsonBuilder.append("\"" + row.getBinary(col.getName()) + "\"");
break;
default:
break;
}
if(columns.hasNext())
jsonBuilder.append(",");
}
jsonBuilder.append("}]}");
return jsonBuilder.toString();
}
}
24 changes: 2 additions & 22 deletions ...du-bundle/nifi-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/PutKudu.java
View file
Open in desktop
Expand Up @@ -41,7 +41,7 @@
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.security.krb.KerberosAction;
import org.apache.nifi.processors.kudu.io.OperationType;
import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
Expand All @@ -51,7 +51,6 @@
import javax.security.auth.login.LoginException;
import java.io.IOException;
import java.io.InputStream;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
Expand Down Expand Up @@ -209,30 +208,11 @@ public void onScheduled(final ProcessContext context) throws IOException, LoginE
createKuduClient(context);
}

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
public void trigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final List<FlowFile> flowFiles = session.get(ffbatch);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't

        if (flowFiles.isEmpty()) {	
            return;	
        }

stay here in PutKudu?

if (flowFiles.isEmpty()) {
return;
}
kerberosUser = getKerberosUser();

final KerberosUser user = kerberosUser;
if (user == null) {
trigger(context, session, flowFiles);
return;
}

final PrivilegedExceptionAction<Void> privelegedAction = () -> {
trigger(context, session, flowFiles);
return null;
};

final KerberosAction<Void> action = new KerberosAction<>(user, privelegedAction, getLogger());
action.execute();
}

private void trigger(final ProcessContext context, final ProcessSession session, final List<FlowFile> flowFiles) throws ProcessException {
final RecordReaderFactory recordReaderFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);

final KuduClient kuduClient = getKuduClient();
Expand Down