influxdata / influxdata/kapacitor
Join ON problem + tolerance
- Dominant language
- Go
- Stars
- 2.4k
- Forks
- 479
- Avg merge
- 4d 16h
- Merged PRs (30d)
- 4
Description
**[Sorry for bad english]**
Here is my tickscript and I am trying to query from two differente measurements.
1. By **first query** getting the historical stats and then take the next monitor label with i would like to use as filter in the last join line.
2. By **second getting** the daily activation data for whole year period and then computing the statistics.
The first Join work fine but the second one based on the first don't.
I can't join the new stream with past stream or it is another problem for example. time tolerance or something else.
JSON file:
```
{
"template-id": "daily_monitor_template",
"dbrps": [{"db": "atsmi", "rp": "autogen"}],
"vars": {
"monitor": {"type": "int", "value": 1},
"statsDB": {"type": "string", "value": "atsmi"},
"statsRP": {"type": "string", "value": "autogen"},
"statsMeasurement": {"type" : "string", "value" : "devices_stats"},
"offsetStats": {"type" : "duration", "value" : "1h" },
"statsPeriod": {"type" : "duration", "value" : "5w"},
"cronStats": {"type" : "string", "value" : "0 3 1 * * *"},
"nextMonitor": {"type" : "string", "value" : "next_monitor"},
"inputDevicesDB": {"type": "string", "value": "atsmi"},
"inputDevicesRP": {"type": "string", "value": "autogen"},
"inputDevicesMeasurement": {"type" : "string", "value" : "devices_daily_activation"},
"dataPeriod": {"type" : "duration", "value" : "50w"},
"groupTime": {"type" : "duration", "value" : "1d"},
"cronData": {"type" : "string", "value" : "0 3 1 * * *" },
"deviceField": {"type" : "string", "value" : "event_counter" },
"offsetData": {"type" : "duration", "value" : "1h" },
"filterWhere": {"type": "lambda", "value": " \"filter_monitor.next_monitor\" == monitor"},
"every": {"type" : "duration", "value" : "15s"},
"deviceFilter": {"type": "lambda", "value": "\"id_device\" == '15576'"},
"statsFilter": {"type": "lambda", "value": "\"next_monitor\" == monitor"}
}
}
```
TICKscipt
```
// Testing variables
var every duration
var deviceFilter = lambda: TRUE
var statsFilter = lambda: TRUE
// COMMON VARIABLES
var monitor int
// decide which kind of data to filter -> depend on task file
// input stats vars
var statsDB string
var statsRP string
var statsMeasurement string
var cronStats string
var statsPeriod duration
var offsetStats duration
var groupStatsBy = [*]
var nextMonitor string
// input device daily data
var inputDevicesDB string
var inputDevicesRP string
var inputDevicesMeasurement string
var cronData string
var dataPeriod duration
var offsetData duration
var groupTime duration
var groupDataBy = *
var deviceField string
var filterWhere lambda
// query the next label from histrical stats data: take the last updated history for device and then filter for label
var filter_monitor = batch
|query('SELECT '+nextMonitor+' FROM "' + statsDB + '"."' + statsRP + '"."' + statsMeasurement + '"')
//.cron(cronStats)
.period(statsPeriod)
.groupBy(groupStatsBy)
// .offset(offsetStats)
.every(every)
.align()
|where(lambda: "next_monitor" == monitor)
|where(lambda: "id_device" == '15576')
|last(nextMonitor)
.as(nextMonitor)
// filter_monitor
// |log()
// // query the daily activation data from database
var device_data = batch
|query('SELECT SUM('+deviceField+') as '+deviceField+' FROM "'+inputDevicesDB+'"."'+inputDevicesRP+'"."'+inputDevicesMeasurement+'"')
//.cron(cronData)
.period(dataPeriod)
.every(every)
.groupBy(time(groupTime), groupDataBy)
.align()
// .offset(offsetData)
|where(deviceFilter)
// Calculation devices stats
var tot_activation = device_data
|sum(deviceField)
.as('sum')
var avg= device_data
|mean(deviceField)
.as('avg')
var std = device_data
|stddev(deviceField)
.as('std')
var min = device_data
|min(deviceField)
.as('min')
var max = device_data
|max(deviceField)
.as('max')
var spread = device_data
|spread(deviceField)
.as('spread')
var median = device_data
|median(deviceField)
.as('median')
var decimo_percentile = device_data
|percentile(deviceField, 10.0)
.as('p10')
var quinto_percentile = device_data
|percentile(deviceField, 5.0)
.as('p5')
var primo_percentile = device_data
|percentile(deviceField, 1.0)
.as('p1')
var device_stats = tot_activation
|join(avg, std, min, max, spread, median, decimo_percentile, quinto_percentile, primo_percentile)
.as('sum','avg','std','min','max','spread','median','p10','p5','p1')
// Does't work with ON, why ?
// .on('id_device')
.tolerance(5m)
.streamName('device_stats')
|eval(lambda: "avg.avg" - "std.std", lambda: "avg.avg" + "std.std")
.as('stdN', 'stdP')
.keep()
|eval(lambda: "sum.sum", lambda: "avg.avg", lambda: "std.std", lambda: "min.min", lambda: "max.max", lambda: "spread.spread", lambda: "median.median", lambda: "p10.p10", lambda: "p5.p5", lambda: "p1.p1")
.as('sum','avg', 'std', 'min', 'max', 'spread', 'median', 'p10', 'p5', 'p1')
var filter_data = device_stats
|join(filter_monitor)
.as('device_stats', 'filter_monitor')
.tolerance(5m)
//Don't work either with ON or tolerance, ??????????????????
// .on('id_device')
|log()
// |log()
// |default()
// .field('filter_monitor.'+nextMonitor, monitor)
// |delete()
// .field('filter_monitor.event_counter')
// |where(lambda: "filter_monitor.next_monitor" == monitor)
```
Contributor guide
Research direction
No source files or tests are named. Reproduce the supplied JSON variables and TICKscript, then compare the working first join with the later joins using ON and tolerance; done means identifying whether the join failure is caused by timestamps, grouping, or unsupported join behavior.
Written by the indexing model from the issue text.
Assessment
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100