Updated trigger example

patch by Mike Adamson; reviewed by Aleksey Yeschenko for CASSANDRA-10257
This commit is contained in:
Mike Adamson 2015-10-20 11:18:37 +01:00 committed by Aleksey Yeschenko
parent d766f4fb20
commit fb463c79e0
4 changed files with 50 additions and 42 deletions

View File

@ -1,4 +1,5 @@
3.0.1
* Updated trigger example (CASSANDRA-10257)
Merged from 2.2:
* (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
Merged from 2.1:

View File

@ -1,11 +1,8 @@
Cassandra Trigger's Example:
=========================
Cassandra Trigger Example:
==========================
InvertedIndex class will create a inverted index of
RowKey:ColumnName:Value to Value:ColumnName:RowKey
NOTE: This example is limited to append-only workloads,
doesn't delete indexes on deletes.
The AuditTrigger class will create a basic audit of
activity on a table.
Installation:
============
@ -13,9 +10,27 @@ change directory to <cassandra_src_dir>/examples/triggers
run "ant jar"
Copy build/trigger-example.jar to <cassandra_conf>/triggers/
Copy conf/* to <cassandra_home>/conf/
Create column family configured in InvertedIndex.properties
Example: Keyspace1.InvertedIndex as configured in InvertedIndex.properties
Configure trigger on the table.
Example: CREATE TRIGGER test1 ON "Keyspace1"."Standard1"
USING 'org.apache.cassandra.triggers.InvertedIndex';
Start inserting data to the column family which has the triggers.
Create the keyspace and table configured in AuditTrigger.properties:
CREATE KEYSPACE test WITH REPLICATION =
{ 'class' : 'SimpleStrategy', 'replication_factor' : '1' };
CREATE TABLE test.audit (key timeuuid, keyspace_name text,
table_name text, primary_key text, PRIMARY KEY(key));
Create a table to add the trigger to:
CREATE TABLE test.test (key text, value text, PRIMARY KEY(key));
Note: The example currently only handles non-composite partition keys
Configure the trigger on the table:
CREATE TRIGGER test1 ON test.test
USING 'org.apache.cassandra.triggers.AuditTrigger';
Start inserting data to the table that has the trigger. For each
partition added to the table an entry should appear in the 'audit' table:
INSERT INTO test.test (key, value) values ('1', '1');
SELECT * FROM test.audit;
key | keyspace_name | primary_key | table_name
--------------------------------------+---------------+-------------+------------
7dc75b60-770f-11e5-9019-033d8af33e6f | test | 1 | test

View File

@ -15,5 +15,5 @@
# specific language governing permissions and limitations
# under the License.
keyspace=Keyspace1
table=InvertedIndex
keyspace=test
table=audit

View File

@ -18,49 +18,42 @@
package org.apache.cassandra.triggers;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Collections;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.Cell;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
public class InvertedIndex implements ITrigger
public class AuditTrigger implements ITrigger
{
private static final Logger logger = LoggerFactory.getLogger(InvertedIndex.class);
private Properties properties = loadProperties();
public Collection<Mutation> augment(ByteBuffer key, ColumnFamily update)
public Collection<Mutation> augment(Partition update)
{
List<Mutation> mutations = new ArrayList<>(update.getColumnCount());
String auditKeyspace = properties.getProperty("keyspace");
String auditTable = properties.getProperty("table");
String indexKeySpace = properties.getProperty("keyspace");
String indexColumnFamily = properties.getProperty("table");
for (Cell cell : update)
{
// Skip the row marker and other empty values, since they lead to an empty key.
if (cell.value().remaining() > 0)
{
Mutation mutation = new Mutation(indexKeySpace, cell.value());
mutation.add(indexColumnFamily, cell.name(), key, System.currentTimeMillis());
mutations.add(mutation);
}
}
RowUpdateBuilder audit = new RowUpdateBuilder(Schema.instance.getCFMetaData(auditKeyspace, auditTable),
FBUtilities.timestampMicros(),
UUIDGen.getTimeUUID());
return mutations;
audit.add("keyspace_name", update.metadata().ksName);
audit.add("table_name", update.metadata().cfName);
audit.add("primary_key", update.metadata().getKeyValidator().getString(update.partitionKey().getKey()));
return Collections.singletonList(audit.build());
}
private static Properties loadProperties()
{
Properties properties = new Properties();
InputStream stream = InvertedIndex.class.getClassLoader().getResourceAsStream("InvertedIndex.properties");
InputStream stream = AuditTrigger.class.getClassLoader().getResourceAsStream("AuditTrigger.properties");
try
{
properties.load(stream);
@ -73,7 +66,6 @@ public class InvertedIndex implements ITrigger
{
FileUtils.closeQuietly(stream);
}
logger.info("loaded property file, InvertedIndex.properties");
return properties;
}
}