<jupyter_start><jupyter_text>Aerospike Query and UDFAerospike queries allow filtering based on a predicate and User Defined Functions (UDFs) offer arbitray server side processing. This notebook illustrates how a query and a UDF can be combined in a useful pattern. Two examples are given: the first with a query with a UDF aggregate function, and the second with a query, predicate expression and a UDF update function. The code for the first example is also available in this [repo](https://github.com/aerospike/aerospike-client-java/blob/master/examples/src/com/aerospike/examples/QuerySum.java).This notebook requires Aerospike datbase running locally and that Java kernel has been installed. Visit [Aerospike notebooks repo](https://github.com/aerospike-examples/interactive-notebooks) for additional details and the docker container. New to notebooks? Here is [one source](https://jupyter-notebook.readthedocs.io/en/stable/examples/Notebook/examples_index.html) to learn more about the Jupyter Notebook.  Ensure database is runningThis notebook requires that Aerospike datbase is running.<jupyter_code>import io.github.spencerpark.ijava.IJava;
import io.github.spencerpark.jupyter.kernel.magic.common.Shell;
IJava.getKernelInstance().getMagics().registerMagics(Shell.class);
%sh asd<jupyter_output><empty_output><jupyter_text>Download Aerospike client from POMHere the Java client version 5.0.0 is installed.<jupyter_code>%%loadFromPOM
<dependencies>
  <dependency>
    <groupId>com.aerospike</groupId>
    <artifactId>aerospike-client</artifactId>
    <version>5.0.0</version>
  </dependency>
</dependencies><jupyter_output><empty_output><jupyter_text>Connect to database and populate test dataThe test data has ten records with user-key 1 through 10, two bins (fields) "binint" and "binstr", in the namespace "test" and set "demo". The two bins are initialized with the user key and a string of the form "(id). (name)".<jupyter_code>import com.aerospike.client.AerospikeClient;
import com.aerospike.client.Bin;
import com.aerospike.client.Key;

AerospikeClient client = new AerospikeClient("localhost", 3000);
System.out.println("Initialized the client and connected to the cluster.");
import com.aerospike.client.policy.ClientPolicy;

String Namespace = "test";
String Set = "demo";
String BinInt = "binint";
String BinStr = "binstr";
int NumRecords = 10;

String Names[] = {"<NAME>", "<NAME>", "<NAME>", "<NAME>", "5. Clark", 
                "<NAME>", "<NAME>", "<NAME>", "<NAME>", "<NAME>"};

ClientPolicy policy = new ClientPolicy();
for (int i = 1; i <= NumRecords; i++) {
    Key key = new Key(Namespace, Set, i);
    Bin bin1 = new Bin(BinInt, i);
    Bin bin2 = new Bin(BinStr, Names[i-1]);
    client.put(policy.writePolicyDefault, key, bin1, bin2);
}

System.out.format("Written %d records in ns=%s set=%s with userkeys 1-%d.",
                  NumRecords, Namespace, Set, NumRecords);<jupyter_output><empty_output><jupyter_text>Create a secondary index<jupyter_code>import com.aerospike.client.policy.Policy;
import com.aerospike.client.query.IndexType;
import com.aerospike.client.task.IndexTask;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.ResultCode;

String IndexName = "idx_numeric_test_demo_binint";

Policy policy = new Policy();
policy.socketTimeout = 0; // Do not timeout on index create.

try {
    IndexTask task = client.createIndex(policy, Namespace, Set, IndexName, 
                                        BinInt, IndexType.NUMERIC);
    task.waitTillComplete();
}
catch (AerospikeException ae) {
    if (ae.getResultCode() != ResultCode.INDEX_ALREADY_EXISTS) {
        throw ae;
    }
}

System.out.format("Created index %s on ns=%s set=%s bin=%s.", 
                                    IndexName, Namespace, Set, BinInt);<jupyter_output><empty_output><jupyter_text>Part 1: Query with aggregate functionIn the first example, we will demonstrate how an aggregate function (sum) is calculated over a stream of records returned by a query. First we will create the aggregate function in an UDF module. A UDF function like a stored procedure that is executed on the server - on all nodes of the cluster. All streams of the partial node-specific answers are then combined locally using the same UDF function. For this reason, the UDF module must be regiateted to the server for the first phase of parallel processing across all node, and also available locally for the final phase of aggregation.  Create UDF module and aggregate functionExamine the following Lua code that is aggregating (reducing) a stream of records into a sum of bin values. Create the UDF module "sum_example.lua" in "udf" directory.   Create a "udf" directory and place sum_example.lua file with this content in it. Alternatively, execute the following two cells to achieve the same effect. -- sum_example.lualocal function reducer(val1,val2)    return val1 + val2endfunction sum_single_bin(stream,name)    local function mapper(rec)        return rec[name]    end    return stream : map(mapper) : reduce(reducer)end<jupyter_code>import java.nio.file.Files;
import java.nio.file.Paths;
import java.io.FileWriter;

void CreateUDFModule(String name, String code) {
    try {
        if (!Files.exists(Paths.get("./udf"))) {
            Files.createDirectory(Paths.get("./udf"));  
        }
        FileWriter fw = new FileWriter("./udf/" + name);
        fw.write(luaCode);
        fw.close();
    }
    catch (Exception e) {
        System.out.format("Failed to create Lua module %s, exception: %s.", 
            "udf/"+name, e);    
    }
}
//  Execute this cell to create UDF module "udf/sum_example.lua" <pre>
// To execute, first convert the cell type from markdown to code.

String luaCode = 
    "-- sum_example.lua" + "\n" +
    "" + "\n" +
    "local function reducer(val1,val2)" + "\n" +
    "    return val1 + val2"  + "\n" +
    "end"  + "\n" +
    "" + "\n" +
    "function sum_single_bin(stream,name)"  + "\n" +
    "    local function mapper(rec)"  + "\n" +
    "        return rec[name]"  + "\n" +
    "    end"  + "\n" +
    "    return stream : map(mapper) : reduce(reducer)"  + "\n" +
    "end";
CreateUDFModule("sum_example.lua", luaCode);

System.out.format("Lua module %s created.", "udf/sum_example.lua"); // </pre><jupyter_output><empty_output><jupyter_text>Register the UDF moduleRegister the lua module for the aggregate function with the server.<jupyter_code>import com.aerospike.client.task.RegisterTask;
import com.aerospike.client.Language;

String UDFDir = "./udf";
String UDFFile = "sum_example.lua";

RegisterTask task = client.register(policy, UDFDir+"/"+UDFFile, UDFFile, Language.LUA);
task.waitTillComplete();

System.out.format("Registered the UDF module %s.", UDFFile);<jupyter_output><empty_output><jupyter_text>Define the query statementThe query statement includes elements such as namespace, set, bins to retrieve, and filter or predicate.<jupyter_code>import com.aerospike.client.query.Statement;
import com.aerospike.client.query.Filter;
import com.aerospike.client.Value;

int begin = 4;
int end = 7;

Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
stmt.setBinNames(BinInt, BinStr);
stmt.setFilter(Filter.range(BinInt, begin, end));
System.out.format("Query on ns=%s set=%s, with bin %s >= %d <= %d",
                  Namespace, Set, BinInt, begin, end);<jupyter_output><empty_output><jupyter_text>Include the aggregate processing and its parameters in the query statement.<jupyter_code>String UDFModule = "sum_example";
String UDFFunction = "sum_single_bin";
stmt.setAggregateFunction(UDFModule, UDFFunction, Value.get(BinInt));

System.out.format("Aggregate function %s added for server processing.", UDFFunction);<jupyter_output><empty_output><jupyter_text>Execute the queryLet's now execute the query.<jupyter_code>import com.aerospike.client.query.ResultSet;

ResultSet rs = client.queryAggregate(null, stmt);
System.out.println("Executed the query with UDF; got results.");<jupyter_output><empty_output><jupyter_text>Process resultsThe expected sum for the records from 4 to 7 (both inclusive) is 4+5+6+7 = 22.<jupyter_code>System.out.println("Processing results:");
try {
    int expected = 22; // 4 + 5 + 6 + 7
    int count = 0;

    while (rs.next()) {
        Object object = rs.getObject();
        long sum;

        if (object instanceof Long) {
            sum = (Long)rs.getObject();
        }
        else {
            System.out.println("Return value not a long: " + object);
            continue;
        }

        if (expected == (int)sum) {
            System.out.format("Sum matched! Value=%d.", expected);
        }
        else {
            System.out.format("Sum mismatch: Expected %d. Received %d.", expected, (int)sum);
        }
        count++;
    }

    if (count == 0) {
        System.out.println("Query failed. No records returned.");
    }
}
finally {
    rs.close();
}<jupyter_output><empty_output><jupyter_text>Part 2: Query, predicate expression, and UDF updateWe will illustrate an update UDF function with a query and predicate expression. Let's say we want to: - update all records by multiplying the integer bin value by 5 - that have the bin value between 2 and 9, - AND whose string bin value have either "Smith" or "Jones" in them. Records with user-keys 3, 4, 6 and 8 meet these conditions.This update can be achieved in different ways using a combination of query, predicate expression, and UDF. For the purpose of this exercise, we use a query with the "between" predicate, a predicate expression for string comparison, and a UDF to update the integer bin. Let's start defining them one by one starting with a new UDF.   Create UDF module with update functionExamine the code below, It simply multiplies a bin value by the input factor and updates the record.  Create a "udf" directory and place update_example.lua file with this content in it. Alternatively, execute the following cell to achieve the same effect. -- update_example.lua function multiplyBy(rec, binName, factor)    rec[binName] = rec[binName] * factor    aerospike:update(rec)end<jupyter_code>// Execute this cell to create UDF module "udf/update_example.lua" <pre>
// To execute, first convert the cell type from markdown to code.

String luaCode = 
    "-- update_example.lua" + "\n" +
    "" + "\n" +
    "function multiplyBy(rec, binName, factor)" + "\n" +
    "    rec[binName] = rec[binName] * factor" + "\n" +
    "    aerospike:update(rec)" + "\n" +
    "end";
CreateUDFModule("update_example.lua", luaCode);

System.out.format("Lua module %s created.", "udf/update_example.lua"); //</pre><jupyter_output><empty_output><jupyter_text>Register the UDF module<jupyter_code>String UDFFile = "update_example.lua";
RegisterTask task = client.register(policy, UDFDir+"/"+UDFFile, 
                                    UDFFile, Language.LUA);
task.waitTillComplete();

System.out.format("Registered the UDF module %s.", UDFFile);<jupyter_output><empty_output><jupyter_text>Define the query statementSpecify the namespace, set, bins, and query filter.<jupyter_code>Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
stmt.setBinNames(BinInt, BinStr);
int begin = 3;
int end = 9;
// Filter is evaluated using a secondary index and therefore can only reference an indexed bin.
stmt.setFilter(Filter.range(BinInt, begin, end));
System.out.format("Query on ns=%s set=%s, with bin %s >= %d <= %d",
                  Namespace, Set, BinInt, begin, end);<jupyter_output><empty_output><jupyter_text>Define the predicate expression filterIn addition to the predicate in the query (which requires a secondary index), additional filtering can be specified using a predicate expression. A predicate expression is specified as part of the request policy and does not require a secondary index. It is evaluated on each record returned after applying the query predicate, and only the records that evaluate True are processed further (in this case for update with the UDF function). Here the predicate expression is the string bin has either "smith" or "jones" in it. We use an expression with an OR clause that combines two regular expression matches.<jupyter_code>// Predicate Expressions are applied on query results on server side.
import com.aerospike.client.policy.WritePolicy;
import com.aerospike.client.exp.Exp;
import com.aerospike.client.query.RegexFlag;

WritePolicy policy = new WritePolicy(client.writePolicyDefault);
policy.filterExp = Exp.build(
    Exp.or(
        Exp.regexCompare(".*smith.*", RegexFlag.ICASE, Exp.stringBin(BinStr)),
        Exp.regexCompare(".*jones.*", RegexFlag.ICASE, Exp.stringBin(BinStr))));

System.out.format("Predicate Expression: (valstr ilike '%%smith%%' || valstr ilike '%%jones%%')");<jupyter_output><empty_output><jupyter_text>Execute the UDF update on filtered records<jupyter_code>// Execute the update UDF function on records that match the statement filter and policy filter. 
// Records are not returned to the client. This asynchronous server call will return 
// before the command is complete. The user can optionally wait for command completion 
// by using the returned ExecuteTask instance.

import com.aerospike.client.task.ExecuteTask;
import com.aerospike.client.Value;

int MultiplicationFactor = 5;
ExecuteTask task = client.execute(policy, stmt, "update_example", "multiplyBy", 
                            Value.get(BinInt), Value.get(MultiplicationFactor));
task.waitTillComplete(3000, 0); // poll time 3s, no timeout

System.out.format("Executed the query and filter expression and applied UDF update to records.");<jupyter_output><empty_output><jupyter_text>View updated records.Remember records 3, 4, 6, and 8 should have received the update, that is, their binint values should be multipled by the specified factor (5).<jupyter_code>import com.aerospike.client.Record;
for (int i = 1; i <= NumRecords; i++) {
    Key key = new Key(Namespace, Set, i);
    Record record = client.get(null, key, BinInt, BinStr);
    System.out.println(record);
}<jupyter_output><empty_output><jupyter_text>Perform clean up<jupyter_code>client.dropIndex(null, Namespace, Set, IndexName);
client.close();
System.out.println("Index dropped and server connection closed.");<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup