Skip to content

Commit 50fae33

Browse files
jnadeau207-collabjnadeau207-collab
committed
under-the-hood: include labeled post ids in reports
Carry logical post IDs through the daily and backfill post-label rows, persist them in optional Thrift fields, aggregate the newest 1,000 distinct IDs per label into monthly rows, and emit them as strings in reportJson alongside postIdsComplete. IDs use the same logical-post identity (initialTweetId.getOrElse(tweetId)) as the existing carried counts, so an edited post chain stays one post in both the count and the ID list. Aggregation keeps the existing map-side combining: UthPostIds.merge is an associative, commutative reduce that sums carried and removed exactly as the previous sum did, and bounds the ID list at every merge. Peak reducer memory per key stays bounded rather than growing with posts per key, and the daily persisted rows carry the same bound as the monthly rows. Field IDs 4 and 11 are intentional, leaving 3 and 10 for the source-provenance fields proposed in #52.
1 parent 85ac72a commit 50fae33

7 files changed

Lines changed: 222 additions & 39 deletions

File tree

‎under-the-hood/scalding/UthDailyPostsBackfillJob.scala‎

Lines changed: 23 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ class UthDailyPostsBackfillApp {
100100
observationMs,
101101
config)
102102
.map {
103-
case (userId, authoredDay, label, asOfDay, carried, removed) =>
103+
case (userId, authoredDay, label, asOfDay, carried, removed, postIds) =>
104104
val age = calendarDaysBetween(authoredDay, asOfDay)
105105
UthDailyPostLabel(
106106
userId = Some(userId),
@@ -111,7 +111,8 @@ class UthDailyPostsBackfillApp {
111111
asOfYyyymmdd = Some(asOfDay),
112112
observationAgeDays = Some(age),
113113
isFinal = Some(age >= config.postObservationDays),
114-
postObservationDays = Some(config.postObservationDays)
114+
postObservationDays = Some(config.postObservationDays),
115+
postIds = Some(postIds)
115116
)
116117
}
117118

@@ -125,7 +126,7 @@ class UthDailyPostsBackfillApp {
125126
rangeEndMs,
126127
config.reducers)
127128
.map {
128-
case (userId, authoredDay, label, carried) =>
129+
case (userId, authoredDay, label, carried, postIds) =>
129130
UthDailyPostLabel(
130131
userId = Some(userId),
131132
authoredYyyymmdd = Some(authoredDay),
@@ -135,7 +136,8 @@ class UthDailyPostsBackfillApp {
135136
asOfYyyymmdd = Some(authoredDay),
136137
observationAgeDays = Some(0),
137138
isFinal = Some(true),
138-
postObservationDays = Some(config.postObservationDays)
139+
postObservationDays = Some(config.postObservationDays),
140+
postIds = Some(postIds)
139141
)
140142
}
141143

@@ -240,7 +242,7 @@ object UthDailyPostsBackfillApp {
240242
rangeEndMs: Long,
241243
observationMs: Long,
242244
config: UthDailyPostsConfig
243-
): TypedPipe[(Long, Int, String, Int, Long, Long)] = {
245+
): TypedPipe[(Long, Int, String, Int, Long, Long, Seq[Long])] = {
244246
val postByTweetId = applyReducers(
245247
posts.map {
246248
case (tweetId, userId, day, logicalId, createdMs) =>
@@ -294,19 +296,29 @@ object UthDailyPostsBackfillApp {
294296

295297
applyReducers(
296298
reduced.collect {
297-
case ((_, userId, day, label, asOfDay, createdMs), (everApply, _, lastApply, lastExpires))
298-
if everApply =>
299+
case (
300+
(logicalId, userId, day, label, asOfDay, createdMs),
301+
(everApply, _, lastApply, lastExpires)
302+
) if everApply =>
299303
val deadline = math.min(createdMs + observationMs, yyyymmddToMs(asOfDay) + DayMs)
300304
val removed =
301305
if (UthDailyPostsApp
302306
.removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L
303307
else 0L
304-
((userId, day, label, asOfDay), (1L, removed))
308+
((userId, day, label, asOfDay), UthPostIds.single(logicalId, removed))
305309
}.group,
306310
config.reducers
307-
).sum.toTypedPipe.map {
308-
case ((userId, day, label, asOfDay), (carried, removed)) =>
309-
(userId, day, label, asOfDay, carried, removed)
311+
).reduce(UthPostIds.merge).toTypedPipe.map {
312+
case ((userId, day, label, asOfDay), summary) =>
313+
(
314+
userId,
315+
day,
316+
label,
317+
asOfDay,
318+
summary.carried,
319+
summary.removed,
320+
summary.postIds
321+
)
310322
}
311323
}
312324
}

‎under-the-hood/scalding/UthDailyPostsJob.scala‎

Lines changed: 27 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ class UthDailyPostsApp {
107107
observationMs,
108108
config)
109109
.map {
110-
case (userId, authoredDay, label, carried, removed) =>
110+
case (userId, authoredDay, label, carried, removed, postIds) =>
111111
val age = calendarDaysBetween(authoredDay, asOfDay)
112112
UthDailyPostLabel(
113113
userId = Some(userId),
@@ -118,7 +118,8 @@ class UthDailyPostsApp {
118118
asOfYyyymmdd = Some(asOfDay),
119119
observationAgeDays = Some(age),
120120
isFinal = Some(age >= config.postObservationDays),
121-
postObservationDays = Some(config.postObservationDays)
121+
postObservationDays = Some(config.postObservationDays),
122+
postIds = Some(postIds)
122123
)
123124
}
124125

@@ -131,7 +132,7 @@ class UthDailyPostsApp {
131132
dayEndMs,
132133
config.reducers)
133134
.map {
134-
case (userId, authoredDay, label, carried) =>
135+
case (userId, authoredDay, label, carried, postIds) =>
135136
UthDailyPostLabel(
136137
userId = Some(userId),
137138
authoredYyyymmdd = Some(authoredDay),
@@ -141,7 +142,8 @@ class UthDailyPostsApp {
141142
asOfYyyymmdd = Some(asOfDay),
142143
observationAgeDays = Some(0),
143144
isFinal = Some(true),
144-
postObservationDays = Some(config.postObservationDays)
145+
postObservationDays = Some(config.postObservationDays),
146+
postIds = Some(postIds)
145147
)
146148
}
147149

@@ -192,7 +194,7 @@ object UthDailyPostsApp {
192194
dayStartMs: Long,
193195
dayEndMs: Long,
194196
reducers: Int
195-
): TypedPipe[(Long, Int, String, Long)] =
197+
): TypedPipe[(Long, Int, String, Long, Seq[Long])] =
196198
if (flagLabels.isEmpty) TypedPipe.empty
197199
else {
198200
val flagged = tweets.flatMap { t =>
@@ -207,11 +209,17 @@ object UthDailyPostsApp {
207209
} else Nil
208210
} else Nil
209211
}
210-
val counts = applyReducers(flagged.group, reducers).sum.keys.map {
211-
case (userId, day, _, label) => ((userId, day, label), 1L)
212-
}.sumByKey
213-
(if (reducers > 0) counts.withReducers(reducers) else counts).toTypedPipe
214-
.map { case ((userId, day, label), carried) => (userId, day, label, carried) }
212+
val distinctPosts = applyReducers(flagged.group, reducers).sum.keys
213+
applyReducers(
214+
distinctPosts.map {
215+
case (userId, day, logicalId, label) =>
216+
((userId, day, label), UthPostIds.single(logicalId, 0L))
217+
}.group,
218+
reducers
219+
).reduce(UthPostIds.merge).toTypedPipe.map {
220+
case ((userId, day, label), summary) =>
221+
(userId, day, label, summary.carried, summary.postIds)
222+
}
215223
}
216224

217225
private[under_the_hood] def loadPosts(
@@ -340,7 +348,7 @@ object UthDailyPostsApp {
340348
dayEndMs: Long,
341349
observationMs: Long,
342350
config: UthDailyPostsConfig
343-
): TypedPipe[(Long, Int, String, Long, Long)] = {
351+
): TypedPipe[(Long, Int, String, Long, Long, Seq[Long])] = {
344352
val postByTweetId = posts.map {
345353
case (tweetId, userId, day, logicalId, createdMs) =>
346354
(tweetId, (logicalId, userId, day, createdMs))
@@ -380,17 +388,19 @@ object UthDailyPostsApp {
380388

381389
applyReducers(
382390
reduced.collect {
383-
case ((_, userId, day, label, createdMs), (everApply, _, lastApply, lastExpires))
384-
if everApply =>
391+
case (
392+
(logicalId, userId, day, label, createdMs),
393+
(everApply, _, lastApply, lastExpires)
394+
) if everApply =>
385395
val deadline = math.min(createdMs + observationMs, dayEndMs)
386396
val removed =
387397
if (removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L else 0L
388-
((userId, day, label), (1L, removed))
398+
((userId, day, label), UthPostIds.single(logicalId, removed))
389399
}.group,
390400
config.reducers
391-
).sum.toTypedPipe.map {
392-
case ((userId, day, label), (carried, removed)) =>
393-
(userId, day, label, carried, removed)
401+
).reduce(UthPostIds.merge).toTypedPipe.map {
402+
case ((userId, day, label), summary) =>
403+
(userId, day, label, summary.carried, summary.removed, summary.postIds)
394404
}
395405
}
396406
}

‎under-the-hood/scalding/UthPostIds.scala‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package com.twitter.visibility.under_the_hood
2+
3+
object UthPostIds {
4+
val MaxPostIdsPerLabel: Int = 1000
5+
6+
final case class Summary(carried: Long, removed: Long, postIds: Seq[Long])
7+
8+
val empty: Summary = Summary(0L, 0L, Vector.empty)
9+
10+
def single(logicalId: Long, removed: Long): Summary =
11+
Summary(1L, if (removed > 0L) 1L else 0L, Vector(logicalId))
12+
13+
def merge(a: Summary, b: Summary): Summary =
14+
Summary(
15+
carried = a.carried + b.carried,
16+
removed = a.removed + b.removed,
17+
postIds = newestDistinct(a.postIds ++ b.postIds, MaxPostIdsPerLabel)
18+
)
19+
20+
def summarize(rows: Iterable[(Long, Long)]): Summary =
21+
rows.foldLeft(empty) {
22+
case (acc, (logicalId, removed)) => merge(acc, single(logicalId, removed))
23+
}
24+
25+
def newestDistinct(postIds: Iterable[Long], limit: Int): Seq[Long] = {
26+
require(limit > 0, s"limit must be > 0; got $limit")
27+
postIds.toSet.toSeq.sorted.takeRight(limit)
28+
}
29+
}

‎under-the-hood/scalding/UthUserMonthMhPublisherJob.scala‎

Lines changed: 28 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,11 @@ object UthUserMonthMhPublisherApp {
165165
label <- row.label
166166
carried <- row.carried
167167
removed <- row.removed
168-
} yield ((userId, monthBucket(day)), (label, dayOfMonth(day), carried, removed))
168+
postIds = row.postIds.getOrElse(Seq.empty)
169+
} yield (
170+
(userId, monthBucket(day)),
171+
(label, dayOfMonth(day), carried, removed, postIds)
172+
)
169173
},
170174
reducers
171175
)
@@ -202,21 +206,34 @@ object UthUserMonthMhPublisherApp {
202206

203207
val postLabelAgg = labelsOpt
204208
.getOrElse(Nil)
205-
.groupBy { case (label, _, _, _) => label }
209+
.groupBy { case (label, _, _, _, _) => label }
206210
.map {
207211
case (label, rows) =>
208-
val days = rows
209-
.groupBy { case (_, day, _, _) => day }
210-
.map {
211-
case (day, dayRows) =>
212-
val best = dayRows.maxBy {
213-
case (_, _, carried, removed) => (carried, removed)
214-
}
215-
UthDayCarriedRemoved(Some(day), Some(best._3), Some(best._4))
212+
val selectedRows = rows
213+
.groupBy { case (_, day, _, _, _) => day }
214+
.values
215+
.map { dayRows =>
216+
dayRows.maxBy {
217+
case (_, _, carried, removed, postIds) =>
218+
(carried, removed, postIds.size)
219+
}
216220
}
217221
.toList
222+
val days = selectedRows
223+
.map {
224+
case (_, day, carried, removed, _) =>
225+
UthDayCarriedRemoved(Some(day), Some(carried), Some(removed))
226+
}
218227
.sortBy(_.dayOfMonth.getOrElse(0))
219-
UthPostLabelAggregate(Some(label), Some(days))
228+
val postIds = UthPostIds.newestDistinct(
229+
selectedRows.flatMap(_._5),
230+
UthPostIds.MaxPostIdsPerLabel
231+
)
232+
UthPostLabelAggregate(
233+
label = Some(label),
234+
days = Some(days),
235+
postIds = Some(postIds)
236+
)
220237
}
221238
.toList
222239
.sortBy(_.label.getOrElse(""))

‎under-the-hood/src/test/scala/com/twitter/visibility/under_the_hood/UthPostIdsSpec.scala‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package com.twitter.visibility.under_the_hood
2+
3+
import org.scalatest.matchers.should.Matchers
4+
import org.scalatest.wordspec.AnyWordSpec
5+
6+
class UthPostIdsSpec extends AnyWordSpec with Matchers {
7+
8+
private def summaryOf(rows: (Long, Long)*): UthPostIds.Summary =
9+
UthPostIds.summarize(rows)
10+
11+
"UthPostIds.single" should {
12+
"count one carried post and normalize any positive removal to one" in {
13+
UthPostIds.single(10L, 0L) shouldBe UthPostIds.Summary(1L, 0L, Vector(10L))
14+
UthPostIds.single(10L, 1L) shouldBe UthPostIds.Summary(1L, 1L, Vector(10L))
15+
UthPostIds.single(10L, 5L).removed shouldBe 1L
16+
}
17+
}
18+
19+
"UthPostIds.merge" should {
20+
"sum carried and removed the way the upstream summed rows" in {
21+
val merged = UthPostIds.merge(UthPostIds.single(1L, 1L), UthPostIds.single(2L, 0L))
22+
merged.carried shouldBe 2L
23+
merged.removed shouldBe 1L
24+
merged.postIds shouldBe Seq(1L, 2L)
25+
}
26+
27+
"treat empty as an identity" in {
28+
val one = UthPostIds.single(7L, 1L)
29+
UthPostIds.merge(UthPostIds.empty, one) shouldBe one
30+
UthPostIds.merge(one, UthPostIds.empty) shouldBe one
31+
}
32+
33+
"be associative and order independent so map-side combining is safe" in {
34+
val a = UthPostIds.single(3L, 0L)
35+
val b = UthPostIds.single(1L, 1L)
36+
val c = UthPostIds.single(2L, 0L)
37+
val left = UthPostIds.merge(UthPostIds.merge(a, b), c)
38+
val right = UthPostIds.merge(a, UthPostIds.merge(b, c))
39+
left shouldBe right
40+
UthPostIds.merge(a, b) shouldBe UthPostIds.merge(b, a)
41+
}
42+
43+
"never let a merged id list exceed the bound" in {
44+
val big = UthPostIds.Summary(
45+
carried = UthPostIds.MaxPostIdsPerLabel.toLong,
46+
removed = 0L,
47+
postIds = (1L to UthPostIds.MaxPostIdsPerLabel.toLong).toVector
48+
)
49+
val other = UthPostIds.Summary(0L, 0L, (5000L to 5100L).toVector)
50+
UthPostIds.merge(big, other).postIds.size shouldBe UthPostIds.MaxPostIdsPerLabel
51+
}
52+
}
53+
54+
"UthPostIds.summarize" should {
55+
"produce ids in deterministic ascending order" in {
56+
summaryOf((30L, 0L), (10L, 0L), (20L, 0L)).postIds shouldBe Seq(10L, 20L, 30L)
57+
}
58+
59+
"reconcile removed counts against carried counts" in {
60+
val s = summaryOf((1L, 1L), (2L, 0L), (3L, 1L))
61+
s.carried shouldBe 3L
62+
s.removed shouldBe 2L
63+
}
64+
65+
"return the empty summary for no rows" in {
66+
UthPostIds.summarize(Nil) shouldBe UthPostIds.Summary(0L, 0L, Vector.empty)
67+
}
68+
69+
"bound the id list while leaving carried counts exact" in {
70+
val rows = (1L to 2500L).map(id => (id, 0L))
71+
val s = UthPostIds.summarize(rows)
72+
s.carried shouldBe 2500L
73+
s.postIds.size shouldBe UthPostIds.MaxPostIdsPerLabel
74+
s.postIds.head shouldBe 1501L
75+
s.postIds.last shouldBe 2500L
76+
}
77+
}
78+
79+
"UthPostIds.newestDistinct" should {
80+
"keep the newest ids by snowflake ordering" in {
81+
UthPostIds.newestDistinct(Seq(5L, 1L, 9L, 3L), 2) shouldBe Seq(5L, 9L)
82+
}
83+
84+
"deduplicate before applying the bound" in {
85+
UthPostIds.newestDistinct(Seq(4L, 4L, 4L, 1L), 10) shouldBe Seq(1L, 4L)
86+
}
87+
88+
"return everything when the bound is not reached" in {
89+
UthPostIds.newestDistinct(Seq(2L, 1L), 10) shouldBe Seq(1L, 2L)
90+
}
91+
92+
"reject a non positive bound" in {
93+
an[IllegalArgumentException] should be thrownBy UthPostIds.newestDistinct(Seq(1L), 0)
94+
}
95+
}
96+
97+
"a per day cap" should {
98+
"not change which ids survive the monthly cap" in {
99+
val dayA = (1L to 2500L).map(id => (id, 0L))
100+
val dayB = (9000L to 9010L).map(id => (id, 0L))
101+
val cappedPerDay =
102+
UthPostIds.summarize(dayA).postIds ++ UthPostIds.summarize(dayB).postIds
103+
val uncapped = (dayA ++ dayB).map(_._1)
104+
UthPostIds.newestDistinct(cappedPerDay, UthPostIds.MaxPostIdsPerLabel) shouldBe
105+
UthPostIds.newestDistinct(uncapped, UthPostIds.MaxPostIdsPerLabel)
106+
}
107+
}
108+
}

‎under-the-hood/strato/columns/underTheHoodReport.User.strato‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,11 +190,14 @@ def buildReportJson(
190190
.flatMap { agg =>
191191
agg.label.filter(underTheHoodLabels.isPostLabel).map { raw =>
192192
val posts = sumCarried(agg.days)
193+
val postIds = agg.postIds.getOrElse(Seq.empty).distinct.sorted
193194
{
194195
label = underTheHoodLabels.postLabelName(raw),
195196
about = underTheHoodLabels.postLabelAbout(raw),
196197
effect = underTheHoodLabels.postLabelEffect(raw),
197198
posts = posts,
199+
postIds = postIds.map(_.toString),
200+
postIdsComplete = postIds.size.toLong == posts,
198201
totalPostsInMonth = postCount,
199202
percentageOfPosts = formatPercentage(posts, postCount),
200203
}

0 commit comments

Comments
 (0)