# Enriching Your Data with Kapacitor **URL:** \ **Category:** Systems\ **Created:** [October 1, 2017, 4:32pm UTC](https://community.influxdata.com/t/enriching-your-data-with-kapacitor/2546 "2017-10-01T16:32:33Z")\ **Posts on this page:** 3\ **Page:** 1
**Author:** ![system](https://us1.discourse-cdn.com/flex023/uploads/influxdata/original/3X/2/f/2feb065bdfd5e4b7728243aa3b9192ad3e8447c1.png) [@system](https://community.influxdata.com/u/system)\ **Post date:** [October 1, 2017, 4:32pm UTC](https://community.influxdata.com/t/enriching-your-data-with-kapacitor/2546/1 "2017-10-01T16:32:33Z")
Originally published at: [Enriching Your Data with Kapacitor | InfluxData](https://www.influxdata.com/blog/enriching-your-data-with-kapacitor/) From time to time, we have seen requests from the community around querying InfluxDB based on a business period such as the typical business day or broken down into shift periods. Consider the following request: How do I summarize my data for the entire month of August just for business hours, defined by Monday through Friday between 0800AM and 0500PM? InfluxQL does not currently possess any functions for filtering based on time. We are limited to: > ``` > SELECT * FROM “mymeasurement” WHERE time >= ‘2017-08-01 08:00:00.000000’ and time <= ‘2017-08-31 17:00:00.000000’; > ``` So how do we accomplish this? The provided Telegraf plugins typically just send a timestamp value, and also have the ability to send some static tags in addition to the metrics and tags associated with the configured plugin. The solution is to use Kapacitor as a “pre-processor” to “decorate” or enrich your data with a computed value that represents a time period that you desire to query. ![](https://us1.discourse-cdn.com/flex023/uploads/influxdata/original/2X/5/5a4e559b55fdb01e6b3f969b9e31180aa671a414.png) For the purposes of this article we are running on `localhost` for Telegraf, InfluxDB, and Kapacitor, but in a full-fledged environment these will be running on different hosts. The first step is to configure Telegraf to write to Kapacitor instead of directly to InfluxDB. In the `[[outputs.influxdb]]` section of your `telegraf.conf` file, there are 3 key settings to consider: > ``` > [[outputs.influxdb]] > urls = [http://localhost:9092] > database = “kap_telegraf” > retention_policy = “autogen” > ``` The urls parameter must point to port 9092 (Kapacitor’s default listening port) instead of port 8086 for InfluxDB. The database parameter should point to a non-existent database (you can ignore Telegraf’s warning about the database not being found). The `retention_policy` parameter should be set to “autogen” or a specific retention policy that you have previously created in your instance. CAUTION: Leaving `retention_policy` set to “” (default) is not the same as “autogen” which is specified as the default retention policy when InfluxDB is initialized. All other settings in `telegraf.conf` can be configured normally for your instance. The next step is to create a TICKscript that will process the data coming from Telegraf. In this example, we are interested in creating a tag that will contain a true or false value if the data point is during business hours that we described above. > ``` > stream > |from() > .database('kap_telegraf') > |eval(lambda: if(((weekday("time") >= 1 AND weekday("time") <= 5) AND (hour("time") >= 8 AND (hour("time")*100+minute("time")) <= 1700)), 'true', 'false')) > .as('business_hours') > .tags('business_hours') > .keep() > |delete() > .field('business_hours') > |influxDBOut() > .database('telegraf') > .retentionPolicy('autogen') > .tag('kapacitor_augmented','true') > ``` In this TICKscript, we are streaming from the non-existent database “kap\_telegraf” that we configured above in our `telegraf.conf` file. The `.from()` method for the `stream()` node only needs the database to match to. We then pass control to an `eval()` node that will evaluate if the point has arrived in the window that we have designed. In this case, we utilize the `weekday()`, `hour()` and `minute()` functions described at [http://docs.influxdata.com/kapacitor/v1.3/tick/expr/#time-functions](http://docs.influxdata.com/kapacitor/v1.3/tick/expr/#time-functions) to evaluate the “time” value. The first part of the condition evaluates whether the day of the week falls between Monday (1) and Friday (5). The second part of the condition evaluates the hour value, being careful to consider that we want to stop at the “00” mark in hour 17 (5pm). To do that, we are multiplying the hour by 100 and adding the result of the `minute()` function to compare to the ending time of 1700. If the `time` value falls within this range, the `eval()` node returns true as a field called `business_hours`. However, since we want to query on this value, we should have it as a tag, so we chain a `.tags()` method to change the value to a tag called `business_hours`. IMPORTANT: The `eval()` node will eliminate all other fields and tags from the stream, so we want to specify `.keep()` to retain these values from the stream. At this point, we have both a field and a tag called `business_hours` that contains the output of the `eval()` node. We should filter this out of the stream by calling a `delete()` node that specifies the field to be deleted via the `.fields()` method. Finally, we pass control in the stream to an `influxDBOut()` node that specifies the destination database and retention policy to write to. We have added an additional static tag for this example called `kapacitor-augmented`. All other data like measurement name is carried through, and it is only necessary to provide new information. In this case, we are going to write to the `telegraf `database and the `autogen `retention policy. Once you have created the TICKscript (referenced below as `businesshours.tick`), we must tell Kapacitor that we want to run it. For this example, we will utilize the Kapacitor CLI to configure the task. > ``` > $ kapacitor define business_hours -type stream -dbrp kap_telegraf.autogen -tick business_hours.tick > ``` This command defines the task called `business_hours` as a stream type listening for writes to `kap_telegraf.autogen` which is the database name and retention policy name that was configured in Telegraf above. Once we have successfully created the task, we need to enable it for processing by Kapacitor. > ``` > $ kapacitor enable business_hours > ``` To show the status, we can ask Kapacitor to list the current tasks: > ``` > $ kapacitor list tasks > ID Type Status Executing Databases and Retention Policies > business_hours stream enabled true ["kap_telegraf"."autogen"] > ``` Once data begins the flow through Kapacitor to InfluxDB, you can then add your condition `AND business_hours=’true’` to the first query we specified: > ``` > SELECT * FROM “mymeasurement” WHERE time >= ‘2017-08-01 08:00:00.000000’ and time <= ‘2017-08-31 17:00:00.000000’ AND business_hours=’true’; > ``` In summary, we have shown a simple example of how to use Kapacitor to enrich your data coming from Telegraf into InfluxDB. This method could be used to add other types of data, potentially obviating the need to create custom Telegraf plugins to meet your business needs. ---
**Author:** ![bcavagnolo](https://avatars.discourse-cdn.com/v4/letter/b/7cd45c/32.png) [@bcavagnolo](https://community.influxdata.com/u/bcavagnolo)\ **Post date:** [October 23, 2017, 6:35pm UTC](https://community.influxdata.com/t/enriching-your-data-with-kapacitor/2546/2 "2017-10-23T18:35:31Z")
Thanks for the post. Can you advise if it is possible to set up authentication between telegraf and kapacitor? ---
**Author:** ![atfujiwara](https://sea1.discourse-cdn.com/flex023/user_avatar/community.influxdata.com/atfujiwara/32/1890_2.png) [@atfujiwara](https://community.influxdata.com/u/atfujiwara)\ **Post date:** [November 21, 2017, 12:15am UTC](https://community.influxdata.com/t/enriching-your-data-with-kapacitor/2546/3 "2017-11-21T00:15:12Z")
I translated this post to Japanese: [Kapacitor でデータを使いやすくする | Yakst](https://yakst.com/ja/posts/4885) If there is any problem about this, please let me know. Thank you.