How to use Aggregation?

1

Hi every body, I need to get all record between 2 value, so I find some example to using UDF like this:

local function range_filter(bin_name, substr_from, substr_to)

   return function(record)
      local val = record[bin_name]
      if type(val) ~= 'string' then
         return false
      end
      if val >= substr_from and val <= substr_to then
         return true
      else
      return false
      end
   end
end

local function rec_to_map(record)

   info(record.bin_names())
   local xrec = map()
   for i, bin_name in ipairs(record.bin_names(record)) do
       xrec[bin_name] = xrec[bin_name]
   end
   return xrec
end

function str_between(stream, bin_name, substr_from, substr_to)

   return stream:filter(range_filter(bin_name, substr_from, substr_to)):map(rec_to_map)
end

My data like this:

±-----------±-----±----±----------+ | datetime | code | age | address | ±-----------±-----±----±----------+ | 1480641211 | 18 | 2 | “80 abc” | | 1480641212 | 76 | 2 | “86 abc” | | 1480641211 | 7 | 2 | “4 abc” | | 1480641211 | 37 | 2 | “26 abc” | | 1480641211 | 91 | 2 | “5 abc” | | 1480641211 | 81 | 5 | “17 abc” | ±-----------±-----±----±----------+

When I try to run code of Go like this:

luaPath, _ := os.Getwd()

luaPath += "/queries/udf/"
as.SetLuaPath(luaPath)
filename := "range_filter"
regTask, err := client.RegisterUDFFromFile(nil, luaPath + filename + ".lua", filename + ".lua", as.LUA)
configs.PanicOnError(err)
configs.PanicOnError(<-regTask.OnComplete())
stm := as.NewStatement(input.DefaultNamespace, input.DefaultSet)
res, err := client.QueryAggregate(nil, stm, filename, "str_between", "age", 1, 100)
configs.PanicOnError(err)
vl := <-res.Results()
fmt.Println(vl);

But rserver serospike trurn data is nil. Some body help me! Thanks

2

First, you could solve this by running a query and filtering your results using a predicate between on a bin. You would have to first build a secondary index on that bin, “age” in your example.

However, if you want to use stream udf, there are a few errors in your stream UDF.

1 - you cannot pass arguments to a filter function. To use arguments passed through your main function, make the filter function local within your calling function. The argument you have in the filter function is the record from the stream.

Here is a fixed up version of your UDF:

local function rec_to_map(rec)
   local xrec = map()
   for i, bin_name in ipairs(record.bin_names(rec)) do
       xrec[bin_name] = rec[bin_name]
   end
   return xrec
end

function str_between(stream, bin_name, substr_from, substr_to)
   local function range_filter(rec)
      local val = rec[bin_name]
      if type(val) == 'string' then
         return false
      end
      if val >= substr_from and val <= substr_to then
         return true
      else
         return false
      end
   end
   return stream:filter(range_filter):map(rec_to_map)
end

Here is sample data and output (I tested via AQL): aql>register module ‘./filterRecords.lua’ OK, 1 module added.

aql>select * from test.demo
+------------+------+-----+----------+
| datetime   | code | age | address  |
+------------+------+-----+----------+
| 1480641211 | 18   | 2   | "80 abc" |
| 1480641211 | 81   | 42  | "17 abc" |
| 1480641211 | 64   | 12  | "29 abc" |
| 1480641211 | 91   | 52  | "5 abc"  |
| 1480641211 | 8    | 85  | "44 abc" |
| 1480641211 | 37   | 20  | "26 abc" |
| 1480641211 | 7    | 8   | "4 abc"  |
| 1480641211 | 76   | 4   | "86 abc" |
| 1480641211 | 54   | 21  | "23 abc" |
+------------+------+-----+----------+
9 rows in set (0.045 secs)

aql>AGGREGATE filterRecords.str_between("age",5,30) ON test.demo
+--------------------------------------------------------------------------+
| str_between                                                              |
+--------------------------------------------------------------------------+
| MAP('{"code":64, "address":"29 abc", "datetime":1480641211, "age":12}')  |
| MAP('{"code":37, "address":"26 abc", "datetime":1480641211, "age":20}')  |
| MAP('{"code":7, "address":"4 abc", "datetime":1480641211, "age":8}')     |
| MAP('{"code":54, "address":"23 abc", "datetime":1480641211, "age":21}')  |
+--------------------------------------------------------------------------+
4 rows in set (0.052 secs)

Hope this helps!

1 Like
3

Realy thanks for your help, I will try it, so if I have some problem when run it, I hope you will help me.

4

hello, @pgupta
I am new to aerospike. And have a question regarding the query that you have posted above. How can i use the above query via node.js client. my requirement is i want to execute a udf that would return multiple records using node.js client.