{"action":"created","issue":{"url":"https://api.github.com/repos/apache/incubator-druid/issues/5693","repository_url":"https://api.github.com/repos/apache/incubator-druid","labels_url":"https://api.github.com/repos/apache/incubator-druid/issues/5693/labels{/name}","comments_url":"https://api.github.com/repos/apache/incubator-druid/issues/5693/comments","events_url":"https://api.github.com/repos/apache/incubator-druid/issues/5693/events","html_url":"https://github.com/apache/incubator-druid/pull/5693","id":317336377,"node_id":"MDExOlB1bGxSZXF1ZXN0MTgzODE5NDc1","number":5693,"title":"support throw duplicate row during realtime ingestion in RealtimePlumber","user":{"login":"kaijianding","id":8663725,"node_id":"MDQ6VXNlcjg2NjM3MjU=","avatar_url":"https://avatars1.githubusercontent.com/u/8663725?v=4","gravatar_id":"","url":"https://api.github.com/users/kaijianding","html_url":"https://github.com/kaijianding","followers_url":"https://api.github.com/users/kaijianding/followers","following_url":"https://api.github.com/users/kaijianding/following{/other_user}","gists_url":"https://api.github.com/users/kaijianding/gists{/gist_id}","starred_url":"https://api.github.com/users/kaijianding/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kaijianding/subscriptions","organizations_url":"https://api.github.com/users/kaijianding/orgs","repos_url":"https://api.github.com/users/kaijianding/repos","events_url":"https://api.github.com/users/kaijianding/events{/privacy}","received_events_url":"https://api.github.com/users/kaijianding/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":{"url":"https://api.github.com/repos/apache/incubator-druid/milestones/25","html_url":"https://github.com/apache/incubator-druid/milestone/25","labels_url":"https://api.github.com/repos/apache/incubator-druid/milestones/25/labels","id":3023864,"node_id":"MDk6TWlsZXN0b25lMzAyMzg2NA==","number":25,"title":"0.13.0","description":"","creator":{"login":"leventov","id":609240,"node_id":"MDQ6VXNlcjYwOTI0MA==","avatar_url":"https://avatars0.githubusercontent.com/u/609240?v=4","gravatar_id":"","url":"https://api.github.com/users/leventov","html_url":"https://github.com/leventov","followers_url":"https://api.github.com/users/leventov/followers","following_url":"https://api.github.com/users/leventov/following{/other_user}","gists_url":"https://api.github.com/users/leventov/gists{/gist_id}","starred_url":"https://api.github.com/users/leventov/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/leventov/subscriptions","organizations_url":"https://api.github.com/users/leventov/orgs","repos_url":"https://api.github.com/users/leventov/repos","events_url":"https://api.github.com/users/leventov/events{/privacy}","received_events_url":"https://api.github.com/users/leventov/received_events","type":"User","site_admin":false},"open_issues":7,"closed_issues":446,"state":"open","created_at":"2018-01-09T18:30:01Z","updated_at":"2018-11-02T04:47:30Z","due_on":null,"closed_at":null},"comments":10,"created_at":"2018-04-24T18:06:58Z","updated_at":"2018-11-04T12:49:19Z","closed_at":"2018-05-04T17:12:25Z","author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/incubator-druid/pulls/5693","html_url":"https://github.com/apache/incubator-druid/pull/5693","diff_url":"https://github.com/apache/incubator-druid/pull/5693.diff","patch_url":"https://github.com/apache/incubator-druid/pull/5693.patch"},"body":"When pulling streaming data, we don't always ingest data without duplicate, like the upper streaming failover and write part of same data again.\r\nSo we add a config to allow to check duplicate row when ingest if we really want more accurate data.\r\n\r\nhow it works:\r\n1. add a dedupColumn config in RealtimeTuningConfig, which is the primary key of the data\r\n2. In RealtimePlumber, sink is bound to an interval, then maintain a Set inside the sink to hold all values of dedupColumn, so dedup doesn't happen across sinks\r\n3. when sink.add(InputRow) called, check the value of dedupColumn of this inputRow whether it is in the Set, if so, reject with return rowCount as -2\r\n4. RealtimeManager increase the ingest/events/duplicate count metric\r\n5. the dedup Set is cleared when the sink is no longer writable:\r\n  5.1 when mergeAndPush which is called at fix rate\r\n  5.2 when got a new sink then try to clearDedupCache for the older sink in case merge process is slow in 5.1\r\n\r\nWhy need a primary key?\r\nIf we compare each columns to distinguish duplication, we need to setup a sink level timeAndDims which costs lots of heap memory.\r\nInstead we ask user to prepare a 'pk' column as dedupColumn (usually, pk is built on several columns like {timestamp, user_id, item_id, order_id}) in their ETL process like in storm, spark, flink, user is glad to do so to gain the ability of dedup, \r\nOf course, 'pk' is no need to be in dimension list, it's only for dedup.\r\n\r\nLimitation:\r\nif the dedupColumn is string type column, whose value maybe md5, use it's long type hashcode (use BKDRHash algorithm) instead to judge whether it is a duplicate row, so there's very small change to throw away a row which is not duplicate. But I think it is worth to do this trade off to reduce heap cost to avoid OOM if maxRowsInMemory is a big number."},"comment":{"url":"https://api.github.com/repos/apache/incubator-druid/issues/comments/435666898","html_url":"https://github.com/apache/incubator-druid/pull/5693#issuecomment-435666898","issue_url":"https://api.github.com/repos/apache/incubator-druid/issues/5693","id":435666898,"node_id":"MDEyOklzc3VlQ29tbWVudDQzNTY2Njg5OA==","user":{"login":"kaijianding","id":8663725,"node_id":"MDQ6VXNlcjg2NjM3MjU=","avatar_url":"https://avatars1.githubusercontent.com/u/8663725?v=4","gravatar_id":"","url":"https://api.github.com/users/kaijianding","html_url":"https://github.com/kaijianding","followers_url":"https://api.github.com/users/kaijianding/followers","following_url":"https://api.github.com/users/kaijianding/following{/other_user}","gists_url":"https://api.github.com/users/kaijianding/gists{/gist_id}","starred_url":"https://api.github.com/users/kaijianding/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kaijianding/subscriptions","organizations_url":"https://api.github.com/users/kaijianding/orgs","repos_url":"https://api.github.com/users/kaijianding/repos","events_url":"https://api.github.com/users/kaijianding/events{/privacy}","received_events_url":"https://api.github.com/users/kaijianding/received_events","type":"User","site_admin":false},"created_at":"2018-11-04T12:49:19Z","updated_at":"2018-11-04T12:49:19Z","author_association":"CONTRIBUTOR","body":"@ankujain this feature is implemented in RealtimePlumber and sink, so it should not care about how firehose is implemented, pull or push"}}
{"actor":{"display_login":"kaijianding"}}