Tag: #apachedruid

ACID and Apache Druid®

Mar 27th, 2023
What is ACID?

ACID refers to a set of properties necessary for a transaction system.

Atomicity – This refers to transactions completing fully or failing without any partial changes. If a bank account is debited then another account needs to be credited. Atomicity means that either both operations succeed or both fail. There should be no scenario in which one succeeds and the other fails.

Consistency – This means that the database should permit only those transactions that comply with the rules. For instance if the rule is that account balance should not be negative and a transaction tries to debit an amount greater than the available balance then it should be rejected. The database must be in a consistent state before and after the transaction.

Isolation – This refers to one transaction not affecting another. If multiple transactions occur concurrently then the end result would be as if they occurred in sequence one after another. If a bank account has a balance of 500 Dollars and two transactions execute concurrently – one debiting 100 Dollars and another crediting 300 Dollars then the result would be a balance of 700 dollars (as if the transactions occurred one after another).
Durability – This means that all committed transactions must be retained even if the system faces a hardware or software failure.

ACID and Druid

Druid is a database for building a modern Analytics application. Druid cannot be used as a system of record and does not support transactions. Given the focus on analytics,ACID guarantees are not fully relevant to Druid. However, it is interesting to consider some of the capabilities of Druid in this light

Atomicity – There are two scenarios in which everything is committed or nothing is committed
1. Batch ingestion – In batch ingestion, the ingestion succeeds and commits data to deep storage only if all rows in the ingestion succeed. Otherwise the entire ingestion fails and no rows are committed.
2. Real time ingestion – Real time ingestion proceeds from a checkpoint for one hour (configurable) and creates another checkpoint and commits data up to that check point. However if any of the rows fail then the ingestion fails in its entirety and begins from the previous checkpoint again.
The above provides sufficient Atomicity guarantee to ensure that Druid does not provide an incorrect picture and that users don’t have to track which data went through and which did not. Taking the example of debit and credit into bank accounts, these records will have same time stamp in the transaction system and when they are ingested into Druid then either both will be ingested or both will not be (with the caveat that both records are part of the same batch ingestion or the same kafka topic). This will ensure that the analytics results are correct.

Consistency – Druid does not allow a lot of rules on the data. There are no primary keys and no foreign keys. If the data types for columns are specified in the ingestion spec then any data that does not conform to the spec will be rejected. When an entity is updated Druid will hold multiple records for that entity as each update is an event in time. Druid provides LATEST sql command to get a consistent picture of the latest state of the entity.

Isolation – Isolation can be seen in Druid’s ability to ingest millions of rows per second while handling 100’s of queries per second at the same time – Historicals handle queries on the data that is already in the system and the indexers respond to queries using the data that is being indexed. Indexers publish the indexed data to deep storage as segments and the historicals download the segments from deep storage and the indexing task is complete only when a Historical has downloaded the segment. Any queries on the data being indexed is handled by the indexer until the Historical has downloaded the segment. This ensures that there are no read-write conflicts (consider the alternate approach of indexers writing the data directly to the historicals. This would have made it very difficult to avoid read-write conflicts). Additionally ,segments (which contain the data indexed and stored in a columnar format) are immutable. This ensures isolation between reads and writes.

Durability – The primary persistent store is the deep storage which is usually a blob storage. So any data that has been committed to deep storage is available even if the entire cluster goes down. A new cluster can be brought using the data in deep storage. Batch ingestion succeeds only if all the data in the batch is committed to deep storage and real time ingestion succeeds only when all data from the previous checkpoint is committed to deep storage.

Conclusion

Druid provides sufficient ACID guarantees from an analytics perspective. Full transactional ACID is not relevant to Druid. The guarantees Druid provides enable building a modern analytics application with large volume data, high throughput real time ingestion and high query concurrency

Links
1. Apache Druid – druid.apache.org
Apache Druid®’s best kept secret – The Tuple Sketch: Part1 – Student’s t test

Feb 2nd, 2023
The tuple sketch is an interesting and intriguing capability in druid. This sketch enables many interesting capabilities
1. Distinct Counts
2. Mean and Variances for different metrics
3. Set operations
4. Student’s t test
In this post I will be looking at the student’s t test.

Student’s t test

The Student’s t test measures the probability of the differences between two distributions being statistically significant. For instance, when a life saving drug is tested and the average life expectancy for the control group is 5 years and the average life expectancy for the group receiving the drug is 6 years then the question is whether this difference is statistically significant or just random?

Student’s t test with Druid

Install Druid and load the wikiticker-2015-09-12-sampled data. The metric “added” in the dataset captures the number of lines added by a user to a channel at a specific time. The difference between average lines added per user for different channels can show the difference in the level of engagement. Run the following query
```
select sum("added") filter (where channel='#en.wikipedia')/
APPROX_COUNT_DISTINCT_DS_THETA("user") filter (where channel='#en.wikipedia') en_avg ,
sum("added") filter (where channel='#hi.wikipedia')/
APPROX_COUNT_DISTINCT_DS_THETA("user") filter (where channel='#hi.wikipedia') hi_avg,
sum("added") filter (where channel='#fr.wikipedia')
/APPROX_COUNT_DISTINCT_DS_THETA("user") filter (where channel='#fr.wikipedia') fr_avg from "wikiticker-2015-09-12-sampled"
```
This give us the below output

The question is whether the difference between avg lines added per user is statistically significant or random. If the difference is significant, then we can conclude that the English wikipedia may have lower engagement than the Hindi wikipedia and the French wikipedia may have a higher engagement than the English wikipedia.The dataset has a number of different metrics (commentLength, lines deleted etc) and each can be analysed to get a different perspective on the engagement level. To understand the statistical significance of the above result I ran a student’s t test using the below query (the tuple sketch uses the native Druid query and has no SQL support for now.
```
{"dataSource": "wikiticker-2015-09-12-sampled","queryType": "groupBy",
  "intervals": [
        "2010-06-27T00:00:00.000Z/2016-06-28T00:00:00.000Z"
    ],
    "granularity":"all",
  "aggregations": [
  {  "type": "filtered",
     "aggregator":
        {
            "type": "arrayOfDoublesSketch",
            "name": "sketch_en",
            "fieldName": "user",
            "metricColumns" : ["added"]
            "nominalEntries": 16384
        },
        "filter": {
       "type": "selector",
       "dimension": "channel",
       "value": "#en.wikipedia",
       "extractionFn": null
     }
     },
     {  "type": "filtered",
        "aggregator":
           {
               "type": "arrayOfDoublesSketch",
               "name": "sketch_hi",
               "fieldName": "user",
               "metricColumns" : ["added"]
               "nominalEntries": 16384
           },
           "filter": {
          "type": "selector",
          "dimension": "channel",
          "value": "#hi.wikipedia",
          "extractionFn": null
        }
        }
],
"postAggregations": [
        {
  "type"  : "arrayOfDoublesSketchTTest",
  "name": "t_test",
  "fields"  : [{"type": "fieldAccess",
            "fieldName": "sketch_en"},
            {"type": "fieldAccess",
                    "fieldName": "sketch_hi"}
            ]
}


],
    "dimensions": [

    ]
}

   
```
This is a groupby query on a druid data source called wikiticker-2015-09-12-sampled for a large time interval (the dataset if just for one day within that interval.

{ “type”: “filtered”,
“aggregator”:
{
“type”: “arrayOfDoublesSketch”,
“name”: “sketch_en”,
“fieldName”: “user”,
“metricColumns” : [“added”]
“nominalEntries”: 16384
},
“filter”: {
“type”: “selector”,
“dimension”: “channel”,
“value”: “#en.wikipedia”,
“extractionFn”: null
}
}

basically creates a tuple sketch (arrayOfDoublesSketch) on user using added as the metric column and filtered for the channel #en.wikipedia. In a similar I have also created a tuple sketch for #hi.wikipedia. The post aggregation

“postAggregations”: [
{
“type” : “arrayOfDoublesSketchTTest”,
“name”: “t_test”,
“fields” : [{“type”: “fieldAccess”,
“fieldName”: “sketch_en”},
{“type”: “fieldAccess”,
“fieldName”: “sketch_hi”}
]
}

]

creates a t test between the two distributions (#en.wikipedia and #hi.wikipedia). When I run this query I get

the column t_test essentially the p value (probability) that the difference in mean between the english and french wikipedia is random and not statistically significant. The p value is between 0 and 1 and a value of 0.07…. above tells me that this difference between the average lines added per user between the hindi and english wikipedia is likely not random and hence statistically significant. Hence one say that there is a difference in the level of engagement between the channels based on the number of lines added per user. The other interesting thing is that the number of distinct users for the hindi wikipedia (16) is much smaller than the number for english (4343). So we get a different picture of the engagement (at least in terms adding content). The point is the tuple sketch allows us to look at both these perspectives in one query that gives us an estimate. When I do a similar exercise between #fr.wikipedia and #en.wikipedia I get

The above shows a p value of 0.52….This tells me that the difference in average lines added per user between english and french wikipedia is statistically not significant.

Key take aways

I have written out a tuple sketch query in druid to do the student’s t test on the wikipedia dataset for different channels to measure the statistical significance of difference in average lines added per user. The tuple sketch allows us to look at both the t-test and the number of distinct users for each channel. This gives us a different picture of the level of engagement in terms of adding content to each channel. The tuple sketch uses an approximation algorithm and hence is very fast on large datasets.

Key links
1. Druid – https://druid.apache.org/docs/latest/tutorials/index.html
2. Druid tuple sketch – https://druid.apache.org/docs/latest/development/extensions-core/datasketches-tuple.html
3. Data sketches – https://datasketches.apache.org/