Duplicates present in data

[p-zach]

I have a job pulling twitter data via a custom fetch plugin. It gathers data by getting a large number of tweets (say, 1000) each minute on a certain topic. If there were fewer than 1000 tweets on that topic in the previous minute, then it will fetch duplicates from the minute before that. I'm relying on mrsm to ignore the duplicates, which has been working until recently, as seen in this image:

I'm confused as to why this is happening, given that all the data fields are the same (not all of them are visible in the image, but further analysis shows that they are in fact identical)
Some of these fields are gotten directly from the Twitter API; some of them are derived based on information in the tweet or user metadata.

Any idea why this is happening or how to fix it?

[bmeares]

Hi @p-zach, thank you for starting a discussion. Normally duplicate rows are automatically removed with the pipe.filter_existing() function. Below is a screenshot of the expected behavior:

Do you still get the same behavior when you explicitly pass your data through filter_existing()? Like you mentioned, usually rows are duplicated when a single value is changed. This function is also skipped with the flags --allow-duplicates or --skip-check-existing, so make sure those flags are not present in your background job.

I hope this helps! Let me know if the issue still persists and we'll dig a bit deeper.

[p-zach]

I do get the same result; filter_existing returns an empty df. I'm not using either of those flags in the job.

[bmeares]

I'm not sure what it might be — can you send an example of the full duplicate rows and a code snippet of the plugin? There might be some concurrency issues at play that we're missing.

[p-zach]

I could not find a way to get a screenshot of all the data together, since it's too big. I know it's not the most helpful, but here's an "example": There are 3 identical rows, all with this data:
| 2022-03-16 20:07:29 | RT @zhang_heqing: #Xinjiang is a wonderful place. #China #AmazingChina https://t.co/2aItR15bvT | Daliakhojely | ['Xinjiang', 'China', 'AmazingChina'] | ['https://twitter.com/zhang_heqing/status/1504161004699402240/video/1'] | en | Khartoum | 1504187618229108739 | 7 | <null> | PhD in Crop Science and Plant Breeding, CAAS, China, Master of Agricultural Sciences and Resource Management in the Tropics and Subtropics (ARTS), @UniBonn. | 2015-04-21 10:21:02 | 83360 | 15.58651 | 32.52318 | 2521 | False |

The basic logic for filling in the data is this:

tweet_data = {
    'timestamp': tweet.created_at,
    'text': tweet.text,
    'username': user.username,
    'urls': None,
    'tags': None,
    'lang': tweet.lang,
    'id': tweet.id,
    'retweet_count': tweet.public_metrics['retweet_count'],
    'user_reported_location': user.location,
    'user_created_at': user.created_at,
    'tweet_count': user.public_metrics['tweet_count'],
    'places_mentioned': None,
    'user_desc': user.description,
    'verified': user.verified,
    'user_loc_lat': None,
    'user_loc_long': None,
    'user_age': (datetime.datetime.now(pytz.utc) - user.created_at).days
}

and there's some additional stuff for geotagging the reported location and getting the URLs, hashtags, and places mentioned.
Would love to provide a screenshot/file of the data but, again, could not figure out how.

[bmeares]

I'm thinking the issue might be a race condition somewhere — are you running multiple syncs on the same pipe concurrently? You can try manually filtering the data through pipe.filter_existing(df) before returning and see if that has any effect, though that is what is supposed to be happening anyway.

To remove your existing duplicates, you'll have to shuffle the data around a bit. Stop the job and run these queries (you can jump into an interactive SQL CLI with mrsm sql):

-- Insert all distinct rows into a temporary table.
SELECT DISTINCT *
INTO tweets_temp
FROM plugin_twitter_tweeper_tweets;

-- Delete the existing rows from the main pipe's table (and keep indices).
TRUNCATE TABLE plugin_twitter_tweeper_tweets;

-- Copy the rows back to the pipe's table.
INSERT INTO plugin_twitter_tweeper_tweets
SELECT * FROM tweets_temp;

-- Once you've verified the rows are there, you can drop the temporary table.
DROP TABLE tweets_temp;

I hope this helps!

[p-zach]

It shouldn't be a race condition, but now that you say that I realize that it could be possible for some combination of long loading times to make that happen... thanks for the advice, I'll try it out later.

[bmeares]

@p-zach Did you get to the root cause of the race condition? My hunch is that if the pipes were being synced regularly by crontab, then a second sync could begin before the first completed. I'll mark this question as answered if the issue has been resolved.

[p-zach]

I didn't get to the root cause; I just ended up running a SQL query to remove the duplicates. It very well may be that a second sync could have begun before the first finished.