· 8 years ago · Apr 11, 2018, 08:40 PM
1package no.uib.sinoa.bergendata.twitter;
2
3import java.io.File;
4import java.io.IOException;
5import java.nio.charset.Charset;
6import java.nio.file.Paths;
7import java.text.SimpleDateFormat;
8
9import org.json.simple.JSONArray;
10import org.json.simple.JSONAware;
11import org.json.simple.JSONObject;
12import org.json.simple.parser.JSONParser;
13import org.json.simple.parser.ParseException;
14
15import com.datastax.driver.core.Session;
16import com.google.common.io.Files;
17
18public class MobDataTweetInserter {
19
20 private CassandraConnector connector;
21 private Session session;
22
23 public Session connect() {
24 connector = new CassandraConnector();
25 connector.connect("18.197.16.119", 9042, "cassandra", "cassandra");
26 return connector.getSession();
27 }
28
29 public void createTables() {
30
31 String ctTwitterTweets = ""
32 + "CREATE TABLE IF NOT EXISTS "
33 + "twitter_tweets ("
34 + " id bigint, "
35 + " id_str text, "
36 + " text text, "
37 + " lang ascii, "
38 + " user_id bigint, "
39 + " created_at timestamp, "
40 + " long_coord float, "
41 + " lat_coord float, "
42 + " long_geo float, "
43 + " lat_geo float, "
44 + " place_id ascii, "
45 + " in_reply_to_status_id_str text, "
46 + " in_reply_to_status_id bigint, "
47 + " in_reply_to_user_id bigint, "
48 + " in_reply_to_user_id_str text, "
49 + " in_reply_to_screen_name text, "
50 + " quoted_id bigint, "
51 + " retweeted_id bigint, "
52 + " retweet_count smallint, "
53 + " retweeted boolean, "
54 + " truncated boolean, "
55 + " is_quote_status boolean, "
56 + " entities_urls list<text>, "
57 + " entities_media list<text>, "
58 + " entities_hashtags list<text>, "
59 + " entities_symbols list<text>, "
60 + " favorite_count smallint, "
61 + " contributors text, "
62 + " source text, "
63 + " possibly_sensitive boolean, "
64 + " favorited boolean, "
65 + " PRIMARY KEY (id, created_at, user_id)"
66 + ") WITH CLUSTERING ORDER BY ("
67 + " created_at DESC, user_id ASC"
68 + ");";
69 session.execute(ctTwitterTweets);
70
71 String ctTwitterUsers = ""
72 + "CREATE TABLE IF NOT EXISTS "
73 + "twitter_users ("
74 + " id bigint, "
75 + " id_str text, "
76 + " utc_offset int, "
77 + " friends_count int, "
78 + " profile_image_url_https text, "
79 + " listed_count smallint, "
80 + " favourites_count int, "
81 + " description text, "
82 + " created_at timestamp, "
83 + " is_translator boolean, "
84 + " protected boolean, "
85 + " screen_name text, "
86 + " is_translation_enabled boolean, "
87 + " translator_type text, "
88 + " geo_enabled boolean, "
89 + " lang ascii, "
90 + " has_extended_profile boolean, "
91 + " verified boolean, "
92 + " profile_image_url text, "
93 + " time_zone text, "
94 + " url text, "
95 + " contributors_enabled boolean, "
96 + " entities_description_urls list<text>, "
97 + " statuses_count int, "
98 + " follow_request_sent boolean, "
99 + " followers_count int, "
100 + " default_profile boolean, "
101 + " following boolean, "
102 + " name text, "
103 + " location text, "
104 + " notifications boolean, "
105 + " PRIMARY KEY (id, followers_count)"
106 + ") WITH CLUSTERING ORDER BY ("
107 + " followers_count DESC"
108 + ");";
109 session.execute(ctTwitterUsers);
110
111 String ctTwitterPlaces = ""
112 + "CREATE TABLE IF NOT EXISTS "
113 + "twitter_places ("
114 + " id ascii, "
115 + " country_code ascii, "
116 + " country text, "
117 + " contained_within list<ascii>, "
118 + " full_name text, "
119 + " bounding_box_type text, "
120 + " bounding_box_long_coords list<float>, "
121 + " bounding_box_lat_coords list<float>, "
122 + " place_type ascii, "
123 + " name text, "
124 + " PRIMARY KEY (id, country, name)"
125 + ") WITH CLUSTERING ORDER BY ("
126 + " country ASC, name ASC"
127 + ");";
128 session.execute(ctTwitterPlaces);
129 }
130
131 final SimpleDateFormat sdfIn = new SimpleDateFormat("EEE MMM dd HH:mm:ss Z yyyy");
132 final SimpleDateFormat sdfOut = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
133
134 public String normaliseTimestamp(String twitter) {
135 try {
136 return sdfOut.format(sdfIn.parse(twitter));
137 } catch (java.text.ParseException e) {
138 e.printStackTrace();
139 return null;
140 }
141 }
142
143 public String escape(String str) {
144 return(str.replace("'", "''"));
145 }
146
147 public void insertLine(String line) {
148 JSONObject jsonTweet = null;
149 try {
150 jsonTweet = (JSONObject)new JSONParser().parse(line);
151
152// Object o = jsonTweet.get("coordinates");
153// if (o instanceof JSONAware)
154// System.out.println(((JSONAware)o).toJSONString());
155// return;
156
157 } catch (ParseException e) {
158 e.printStackTrace();
159 return;
160 }
161
162 // set user_id of twitter_tweet
163 jsonTweet.put("user_id", ((JSONObject)jsonTweet.get("user")).get("id"));
164
165 // set place of twitter_tweet if present
166 if (jsonTweet.containsKey("place") && null != jsonTweet.get("place")) {
167 jsonTweet.put("place_id", ((JSONObject)jsonTweet.get("place")).get("id"));
168 }
169
170 // set quoted_id of twitter_tweet if present
171 if (jsonTweet.containsKey("quoted_status")) {
172 jsonTweet.put("quoted_id", ((JSONObject)jsonTweet.get("quoted_status")).get("id"));
173 }
174
175 // set retweeted_id of twitter_tweet if present
176 if (jsonTweet.containsKey("retweeted_status")) {
177 jsonTweet.put("retweeted_id", ((JSONObject)jsonTweet.get("retweeted_status")).get("id"));
178 }
179
180 // flatten coordinates of twitter_tweet
181 if (jsonTweet.get("coordinates") != null) {
182 JSONObject coordinates = ((JSONObject)jsonTweet.get("coordinates"));
183 jsonTweet.put("long_coord", ((JSONArray)coordinates.get("coordinates")).get(0));
184 jsonTweet.put("lat_coord", ((JSONArray)coordinates.get("coordinates")).get(1));
185 }
186
187 // flatten geo of twitter_tweet
188 if (jsonTweet.get("geo") != null) {
189 JSONObject geo = ((JSONObject)jsonTweet.get("geo"));
190 jsonTweet.put("long_geo", ((JSONArray)geo.get("coordinates")).get(0));
191 jsonTweet.put("lat_geo", ((JSONArray)geo.get("coordinates")).get(1));
192 }
193
194 // flatten entities of twitter_tweet
195 if (jsonTweet.containsKey("entities")) {
196 JSONObject entities = ((JSONObject)jsonTweet.get("entities"));
197
198 if (entities.containsKey("urls")) {
199 JSONArray urls = new JSONArray();
200 for (Object url : ((JSONArray)entities.get("urls")).toArray())
201 urls.add(((JSONObject) url).get("url"));
202 jsonTweet.put("entities_urls", urls);
203 }
204
205 if (entities.containsKey("media")) {
206 JSONArray media = new JSONArray();
207 for (Object medium : ((JSONArray)entities.get("media")).toArray())
208 media.add(((JSONObject) medium).get("url"));
209 jsonTweet.put("entities_urls", media);
210 }
211
212 if (entities.containsKey("hashtags")) {
213 JSONArray entities_hashtags = new JSONArray();
214 for (Object hashtag : ((JSONArray)entities.get("hashtags")).toArray())
215 entities_hashtags.add(((JSONObject) hashtag).get("text"));
216 jsonTweet.put("entities_hashtags", entities_hashtags);
217 }
218
219 if (entities.containsKey("symbols"))
220 jsonTweet.put("entities_symbols", entities.get("symbols"));
221 }
222
223 // separate user from twitter_tweet
224 JSONObject jsonUser = (JSONObject)jsonTweet.get("user");
225
226 // flatten entities of twitter_user
227 if (jsonTweet.containsKey("entities") && ((JSONObject)jsonTweet.get("entities")).containsKey("description")) {
228 jsonTweet.put("entities_description_urls", ((JSONObject)((JSONObject)jsonTweet.get("entities")).get("description")).get("urls"));
229 }
230
231 // separate place from twitter_tweet
232 JSONObject jsonPlace = (JSONObject)jsonTweet.get("place");
233
234 // flatten bounding_box of twitter place
235 if (jsonPlace != null && jsonPlace.containsKey("bounding_box")) {
236 JSONObject bounding_box = ((JSONObject)jsonPlace.get("bounding_box"));
237
238 if (bounding_box.containsKey("coordinates")) {
239 Object coordinates = bounding_box.get("coordinates");
240
241 System.out.println(((JSONAware)bounding_box.get("coordinates")).toJSONString());
242
243 if (coordinates instanceof JSONObject) {
244
245 if (((JSONArray)((JSONObject)coordinates).get("coordinates")).get(0) instanceof JSONArray)
246 coordinates = ((JSONObject)coordinates).get("coordinates");
247 else
248 coordinates = new JSONArray().add(((JSONObject)coordinates).get("coordinates"));
249
250 } else {
251
252 if (((JSONArray)coordinates).get(0) instanceof JSONArray)
253 coordinates = ((JSONArray)coordinates).get(0);
254 else
255 coordinates = new JSONArray().add(coordinates);
256
257 }
258
259 JSONArray long_coords = new JSONArray();
260 JSONArray lat_coords = new JSONArray();
261 for (Object coordinate : ((JSONArray)coordinates).toArray()) {
262 long_coords.add(((JSONArray) coordinate).get(0));
263 long_coords.add(((JSONArray) coordinate).get(1));
264 }
265 jsonPlace.put("bounding_box_long_coords", long_coords);
266 jsonPlace.put("bounding_box_lat_coords", lat_coords);
267 }
268
269 if (bounding_box.containsKey("type"))
270 jsonPlace.put("bounding_box_type", bounding_box.get("type"));
271 }
272
273 // remove unused objects in twitter_tweet
274 jsonTweet.remove("user");
275 jsonTweet.remove("metadata");
276 jsonTweet.remove("coordinates");
277 jsonTweet.remove("geo");
278 jsonTweet.remove("place");
279 jsonTweet.remove("quoted_status");
280 jsonTweet.remove("quoted_status_id");
281 jsonTweet.remove("quoted_status_id_str");
282 jsonTweet.remove("retweeted_status");
283 jsonTweet.remove("entities");
284 jsonTweet.remove("extended_entities");
285
286 // remove unused objects in twitter_user
287 jsonUser.remove("entities");
288 jsonUser.remove("extended_entities");
289 jsonUser.remove("profile_image_url_https");
290 jsonUser.remove("profile_background_image_url");
291 jsonUser.remove("default_profile_image");
292 jsonUser.remove("profile_background_image_url_https");
293 jsonUser.remove("profile_link_color");
294 jsonUser.remove("profile_background_color");
295 jsonUser.remove("profile_sid)ebar_border_color");
296 jsonUser.remove("profile_text_color");
297 jsonUser.remove("profile_image_url");
298 jsonUser.remove("profile_background_tile");
299 jsonUser.remove("profile_banner_url");
300 jsonUser.remove("profile_use_background_image");
301 jsonUser.remove("profile_sidebar_fill_color");
302 jsonUser.remove("profile_sidebar_border_color");
303
304 // remove unused objects in place
305 if (jsonPlace != null) {
306 jsonPlace.remove("bounding_box");
307 jsonPlace.remove("attributes");
308 jsonPlace.remove("url");
309 }
310
311 // normalise created_at timestamps
312 jsonTweet.put("created_at", normaliseTimestamp((String)jsonTweet.get("created_at")));
313 jsonUser.put("created_at", normaliseTimestamp((String)jsonUser.get("created_at")));
314
315 // insert into Cassandra
316 session.execute("INSERT INTO twitter_tweets JSON '" + escape(jsonTweet.toJSONString()) + "' ; ");
317 session.execute("INSERT INTO twitter_users JSON '" + escape(jsonUser.toJSONString()) + "' ; ");
318 if (jsonPlace != null)
319 session.execute("INSERT INTO twitter_places JSON '" + escape(jsonPlace.toJSONString()) + "' ; ");
320 }
321
322 public void insertFile(File file) {
323 try {
324 for (String line : Files.readLines(file, Charset.forName("UTF-8")))
325 insertLine(line);
326 } catch (IOException e) {
327 e.printStackTrace();
328 }
329 }
330
331 public void insertData() {
332 File dataFolder = Paths.get("data/").toFile();
333 for (File file : dataFolder.listFiles()) {
334 insertFile(file);
335 }
336 }
337
338 public MobDataTweetInserter() {
339
340 session = connect();
341
342 session.execute("USE bergendata;");
343 session.execute("DROP TABLE IF EXISTS twitter_tweets ; ");
344 session.execute("DROP TABLE IF EXISTS twitter_users ; ");
345 session.execute("DROP TABLE IF EXISTS twitter_places ; ");
346
347 createTables();
348
349 insertData();
350
351 connector.close();
352 }
353
354 public static void main(String[] args) {
355 new MobDataTweetInserter();
356 }
357}