Skip to content

Commit 8dead02

Browse files
anourianouri
authored
Merge pull request #96 from IBMStreams/kerberos
pull request from kerberos to master branch
2 parents c4ef088 + 2f3df45 commit 8dead02

File tree

31 files changed

+676
-444
lines changed

31 files changed

+676
-444
lines changed

‎README.md

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -39,17 +39,10 @@ The following applications in the `samples` directory are good starting points:
3939
## Building the toolkit
4040
The toolkit uses [Maven](http://maven.apache.org/) to download the needed dependencies.
4141
* Set M2_HOME to point to the maven directory.
42-
* Pick the correct pom file for your install. In com.ibm.streamsx.hbase, there are three example pom files:
43-
44-
* pom-v094.xml: HBase 0.94, hadoop 1
45-
* pom-v096-hadoop1.xml, HBase 0.96, hadoop 1
46-
* pom-v096-hadoop2.xml, HBase 0.96, hadoop 2
47-
* pom-v1.2-hadoop2.xml, HBase 1.2, hadoop 2.7
48-
49-
Copy the correct file for your HBase and Hadoop install to pom.xml
42+
* The pom.xml file has ‘exclusion’ section and download only needed jar libraries from apache.org.repositories.
43+
* It is possible to change the pom.xml file to download another version of hadoop or hbase jar libraries.
5044
* Run `ant` at the top level. This will build the toolkit, but also download all the necessary jars into `opt/downloaded`. These jars are used at toolkit build time, but also at the toolkit runtime.
5145

52-
5346
## Troubleshooting
5447
Please enter an issue on GitHub for defects and other problems.
5548

‎build.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@
126126
destfile="${releasefilename}"
127127
basedir="${basedir}"
128128
includes="${tkinfo.info:identity.info:name}/** samples/**"
129-
excludes="**/.gitignore **/.settings/** **/.settings **/.project **/.classpath **/opt/**"
129+
excludes="**/.gitignore **/.settings/** **/.settings **/.project **/.classpath"
130130
/>
131131
<checksum file="${releasefilename}"/>
132132
<checksum algorithm="sha1" file="${releasefilename}"/>

‎com.ibm.streamsx.hbase/.classpath

Lines changed: 26 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,32 @@
11
<?xml version="1.0" encoding="UTF-8"?>
22
<classpath>
3-
<classpathentry kind="src" output="impl/java/bin" path="impl/java/src"/>
3+
<classpathentry kind="src" output="opt/java/bin" path="opt/java/src"/>
44
<classpathentry exported="true" kind="con" path="com.ibm.streams.java/com.ibm.streams.operator"/>
55
<classpathentry exported="true" kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
6+
<classpathentry kind="lib" path="opt/downloaded/commons-cli-1.2.jar"/>
7+
<classpathentry kind="lib" path="opt/downloaded/commons-codec-1.4.jar"/>
8+
<classpathentry kind="lib" path="opt/downloaded/commons-collections-3.2.2.jar"/>
69
<classpathentry kind="lib" path="opt/downloaded/commons-configuration-1.6.jar"/>
7-
<classpathentry kind="lib" path="opt/downloaded/commons-logging-1.1.1.jar"/>
8-
<classpathentry kind="lib" path="opt/downloaded/hadoop-auth-2.2.0.jar"/>
9-
<classpathentry kind="lib" path="opt/downloaded/hadoop-common-2.2.0.jar"/>
10-
<classpathentry kind="lib" path="opt/downloaded/htrace-core-2.04.jar"/>
11-
<classpathentry kind="lib" path="opt/downloaded/hbase-client-0.98.4-hadoop2.jar"/>
12-
<classpathentry kind="lib" path="opt/downloaded/hbase-common-0.98.4-hadoop2.jar"/>
13-
<classpathentry kind="lib" path="opt/downloaded/hbase-protocol-0.98.4-hadoop2.jar"/>
14-
<classpathentry kind="output" path="impl/java/bin"/>
10+
<classpathentry kind="lib" path="opt/downloaded/commons-io-2.4.jar"/>
11+
<classpathentry kind="lib" path="opt/downloaded/commons-lang-2.6.jar"/>
12+
<classpathentry kind="lib" path="opt/downloaded/commons-logging-1.1.3.jar"/>
13+
<classpathentry kind="lib" path="opt/downloaded/guava-11.0.2.jar"/>
14+
<classpathentry kind="lib" path="opt/downloaded/hadoop-annotations-2.7.3.jar"/>
15+
<classpathentry kind="lib" path="opt/downloaded/hadoop-auth-2.7.3.jar"/>
16+
<classpathentry kind="lib" path="opt/downloaded/hadoop-common-2.7.3.jar"/>
17+
<classpathentry kind="lib" path="opt/downloaded/hbase-annotations-1.2.4.jar"/>
18+
<classpathentry kind="lib" path="opt/downloaded/hbase-client-1.2.4.jar"/>
19+
<classpathentry kind="lib" path="opt/downloaded/hbase-common-1.2.4.jar"/>
20+
<classpathentry kind="lib" path="opt/downloaded/hbase-protocol-1.2.4.jar"/>
21+
<classpathentry kind="lib" path="opt/downloaded/htrace-core-3.1.0-incubating.jar"/>
22+
<classpathentry kind="lib" path="opt/downloaded/javaee-api-7.0.jar"/>
23+
<classpathentry kind="lib" path="opt/downloaded/log4j-1.2.15.jar"/>
24+
<classpathentry kind="lib" path="opt/downloaded/metrics-core-2.2.0.jar"/>
25+
<classpathentry kind="lib" path="opt/downloaded/netty-all-4.0.52.Final.jar"/>
26+
<classpathentry kind="lib" path="opt/downloaded/protobuf-java-2.5.0.jar"/>
27+
<classpathentry kind="lib" path="opt/downloaded/servlet-api-2.5.jar"/>
28+
<classpathentry kind="lib" path="opt/downloaded/slf4j-api-1.7.10.jar"/>
29+
<classpathentry kind="lib" path="opt/downloaded/slf4j-log4j12-1.7.10.jar"/>
30+
<classpathentry kind="lib" path="opt/downloaded/zookeeper-3.4.6.jar"/>
31+
<classpathentry kind="output" path="opt/java/bin"/>
1532
</classpath>

‎com.ibm.streamsx.hbase/README renamed to ‎com.ibm.streamsx.hbase/README.md

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,29 +2,47 @@ This toolkit is for interacting with Apache HBASE. It assumes Apache HBASE
22
is already installed and running.
33

44
To use the operators, you must set HADOOP_HOME and HBASE_HOME in your
5-
environment before compiling.
5+
environment before compiling.
66

77
This toolkit provides Streams operators to work with the HBASE API,
88
so the operators are named after the HBASE API calls. There is another
99
toolkit for HBASE that takes a more database-like approach.
1010

1111
Operators:
12-
HBASEPut: Put a tuple into HBASE. The row and value must be supplied in the tuple.
12+
13+
**HBASEPut**: Put a tuple into HBASE. The row and value must be supplied in the tuple.
1314
The columnFamily and columnQualifier can be either from operator parameters or from
1415
the tuple. To use checkAndPut, the incoming tuple needs to include an attribute
1516
specifying the check to perform. See the example for details.
16-
HBASEGet: Get a tuple from HBASE. The row must be specified in the input tuple.
17+
18+
**HBASEGet**: Get a tuple from HBASE. The row must be specified in the input tuple.
1719
The columnFamily and columnQualifier may be in the input tuple, operator parameters,
1820
or unspecified. The values from HBASE are placed in an attribute of the output
1921
tuple. That attribute must be either a string, a long, or a map. See examples for details.
20-
HBASEDelete: Delete a tuple from HBASE. The row must be supplied in the tuple.
22+
23+
**HBASEDelete**: Delete a tuple from HBASE. The row must be supplied in the tuple.
2124
The columnFamily and columnQualifier can either be operator parameters or from the tuple.
2225
To use use checkAndDelete, the incoming tuple must include and attribute specifying the
2326
check to perform. See the example for details.
24-
HBASEIncrement: Increment a value in HBASE. The row must be supplied in the input tuple.
27+
28+
**HBASEIncrement**: Increment a value in HBASE. The row must be supplied in the input tuple.
2529
The columnFamily and columnQualifier can be from either the operator parameters or from
2630
the input tuple. The increment value can be in the tuple, and operator parameter,
2731
or unspecified. If unspecified, the increment value uses the default.
28-
HBASEScan: Scan a table (ie, get all tuples in the table). The output can be limited to
32+
33+
**HBASEScan**: Scan a table (ie, get all tuples in the table). The output can be limited to
2934
specific column families or to a column family, column qualifier pair. A start row
30-
and an end row may also be specified.
35+
and an end row may also be specified.
36+
37+
**Kerberos Authentication**
38+
The streamsx.hbase toolkit support from version 3.1.0 kerberos authentication
39+
Kerberos authentication is a network protocol to provide strong authentication for client/server applications.
40+
41+
If not done already, enable Kerberos authentication on your Hadoop cluster using the following links to enable the kerberos authentication.
42+
https://hortonworks.com/blog/ambari-kerberos-support-hbase-1/
43+
https://www.cloudera.com/documentation/enterprise/5-7-x/topics/cdh_sg_hbase_authentication.html
44+
https://www.ibm.com/support/knowledgecenter/en/SSPT3X_4.2.0/com.ibm.swg.im.infosphere.biginsights.admin.doc/doc/admin_iop_kerberos.html
45+
46+
After enabling the Kerberos authentication, copy the hbase server keytab and hbase configuration file "hbase-site.xml" from hadoop server into your IBM Streams server in a directory and use them in your SPL application.
47+
48+

‎com.ibm.streamsx.hbase/build.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@
102102

103103
<!-- Downloads libraries using maven -->
104104
<target name="maven-deps" depends="init,nomaven" if="env.M2_HOME">
105-
<fail unless="pomset" message="There must be a pom.xml file in com.ibm.streamsx.hbase in order to use maven to download necessary dependencies. Three sample pom files are included: pom-v094.xml (hbase 0.94, hadoop 1),pom-v096-hadoop1.xml (hbase 0.96, hadoop 1), and pom-v096-hadoop2.xml (hbase 0.96, hadoop 2). Please rename the correct one for your install to pom.xml"/>
105+
<fail unless="pomset" message="There must be a pom.xml file in com.ibm.streamsx.hbase in order to use maven to download necessary dependencies."/>
106106
<exec executable="${maven.bin}" failonerror="true">
107107
<arg value="dependency:copy-dependencies"/>
108108
<arg value="-DoutputDirectory=${ext.downloads.dir}"/>

‎com.ibm.streamsx.hbase/impl/java/src/com/ibm/streamsx/hbase/HBASEDelete.java

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
1-
/* Copyright (C) 2013-2014, International Business Machines Corporation */
2-
/* All Rights Reserved */
1+
/* Copyright (C) 2013-2018, International Business Machines Corporation */
2+
/* All Rights Reserved */
33

44
package com.ibm.streamsx.hbase;
55

@@ -9,15 +9,14 @@
99
import java.util.Set;
1010

1111
import org.apache.hadoop.hbase.client.Delete;
12-
import org.apache.hadoop.hbase.client.HTableInterface;
12+
import org.apache.hadoop.hbase.client.Table;
1313
import org.apache.log4j.Logger;
1414

1515
import com.ibm.streams.operator.OperatorContext;
1616
import com.ibm.streams.operator.OperatorContext.ContextCheck;
1717
import com.ibm.streams.operator.StreamingInput;
1818
import com.ibm.streams.operator.Tuple;
1919
import com.ibm.streams.operator.compile.OperatorContextChecker;
20-
import com.ibm.streams.operator.logging.LoggerNames;
2120
import com.ibm.streams.operator.model.Icons;
2221
import com.ibm.streams.operator.model.InputPortSet;
2322
import com.ibm.streams.operator.model.InputPortSet.WindowMode;
@@ -35,7 +34,7 @@
3534
* HBASE table. .
3635
* <P>
3736
*/
38-
@PrimitiveOperator(name = "HBASEDelete", namespace = "com.ibm.streamsx.hbase", description = "The `HBASEDelete` operator deletes an entry, an entire row, a columnFamily in a row, or a columnFamily, columnQualifier pair in a row from HBASE. It can also optionally do a checkAndDelete operation."
37+
@PrimitiveOperator(name = "HBASEDelete", namespace = "com.ibm.streamsx.hbase", description = "The `HBASEDelete` operator deletes an entry, an entire row, a columnFamily in a row, or a columnFamily, columnQualifier pair in a row from an HBase table. It can also optionally do a checkAndDelete operation."
3938
+ HBASEOperator.DOC_BLANKLINE
4039
+ "The behavior of the operator depends on its parameters:"
4140
+HBASEOperator.DOC_BLANKLINE
@@ -180,20 +179,20 @@ public synchronized void initialize(OperatorContext context)
180179
@Override
181180
public void process(StreamingInput<Tuple> stream, Tuple tuple)
182181
throws Exception {
183-
HTableInterface myTable = connection.getTable(tableNameBytes);
182+
Table myTable = getHTable();
184183
byte row[] = getRow(tuple);
185184
Delete myDelete = new Delete(row);
186185

187186
if (DeleteMode.COLUMN_FAMILY == deleteMode) {
188187
byte colF[] = getColumnFamily(tuple);
189-
myDelete.deleteFamily(colF);
188+
myDelete.addFamily(colF);
190189
} else if (DeleteMode.COLUMN == deleteMode) {
191190
byte colF[] = getColumnFamily(tuple);
192191
byte colQ[] = getColumnQualifier(tuple);
193192
if (deleteAll) {
194-
myDelete.deleteColumns(colF, colQ);
193+
myDelete.addColumns(colF, colQ);
195194
} else {
196-
myDelete.deleteColumn(colF, colQ);
195+
myDelete.addColumn(colF, colQ);
197196
}
198197
}
199198

@@ -237,7 +236,8 @@ public void process(StreamingInput<Tuple> stream, Tuple tuple)
237236
@Override
238237
protected void flushBuffer() throws IOException {
239238
if (connection != null && !connection.isClosed()) {
240-
HTableInterface myTable = connection.getTable(tableNameBytes);
239+
// HTableInterface myTable = connection.getTable(tableNameBytes);
240+
Table myTable =getHTable();
241241
if (myTable != null && deleteList != null && deleteList.size() > 0) {
242242
synchronized (listLock) {
243243
if (deleteList != null && deleteList.size() > 0) {

‎com.ibm.streamsx.hbase/impl/java/src/com/ibm/streamsx/hbase/HBASEGet.java

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
1-
/* Copyright (C) 2013-2014, International Business Machines Corporation */
2-
/* All Rights Reserved */
1+
/* Copyright (C) 2013-2018, International Business Machines Corporation */
2+
/* All Rights Reserved */
33

44
package com.ibm.streamsx.hbase;
55

@@ -10,10 +10,9 @@
1010
import java.util.Set;
1111

1212
import org.apache.hadoop.hbase.client.Get;
13-
import org.apache.hadoop.hbase.client.HTableInterface;
13+
import org.apache.hadoop.hbase.client.Table;
1414
import org.apache.hadoop.hbase.client.Result;
1515
import org.apache.log4j.Logger;
16-
1716
import com.ibm.streams.operator.Attribute;
1817
import com.ibm.streams.operator.OperatorContext;
1918
import com.ibm.streams.operator.OperatorContext.ContextCheck;
@@ -57,7 +56,7 @@
5756
* columnQualifers to values.
5857
*
5958
*/
60-
@PrimitiveOperator(name = "HBASEGet", namespace = "com.ibm.streamsx.hbase", description = "The `HBASEGet` operator gets tuples from HBASE. It is similar to the `ODBCEnrich` operator in the Database Toolkit. It puts the result in the attribute of the output port that is specified in the "
59+
@PrimitiveOperator(name = "HBASEGet", namespace = "com.ibm.streamsx.hbase", description = "The `HBASEGet` operator gets tuples from an HBase table. It is similar to the `ODBCEnrich` operator in the Database Toolkit. It puts the result in the attribute of the output port that is specified in the "
6160
+ HBASEGet.OUT_PARAM_NAME
6261
+ " parameter. The operator accepts four types of queries. In the simplest case, you specify a row, columnFamily, and columnQualifier, and the output value is the single value in that entry. \\n"
6362
+" stream<rstring who, rstring infoType, rstring requestedDetail, rstring value, \\n"
@@ -131,7 +130,6 @@ private enum OutputMode {
131130
private static final String defaultOutAttrName = "value";
132131
private String outAttrName = defaultOutAttrName;;
133132
private String successAttr = null;
134-
private SingleOutputMapper primativeOutputMapper = null;
135133

136134
@Parameter(name = SUCCESS_PARAM_NAME, description = "This parameter specifies the name of attribute of the output port where the operator puts a count of the values it populated.", optional = true)
137135
public void setSuccessAttr(String name) {
@@ -319,7 +317,7 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple)
319317
myGet.addFamily(colF);
320318
}
321319
}
322-
HTableInterface myTable = connection.getTable(tableNameBytes);
320+
Table myTable = getHTable();
323321
Result r = myTable.get(myGet);
324322

325323
int numResults = r.size();

0 commit comments

Comments
 (0)