Big Data Concept

The report was built correctly. The metrics were arranged so that they can be read back at a glance, they were tested for whether they can be gamed, the narrative carried the steps from finding to recommendation without gaps, and the items that must be pinned down so that the report can be produced a second time were counted. Underneath all of this work sits a single unwritten assumption: the dataset fits on one machine. The source is opened, the rows are taken into memory, the computation is done, and the result is written.

This topic removes that assumption. The first question is the crudest one: when is a dataset considered “too big for a single machine”? The answer is not an adjective but a number, and without writing down the decisions that produce that number, the number does not exist. The lesson builds volume, velocity, and variety not as three separate qualitative claims but as six countable units: bytes and rows, rows/period and latency, the number of distinct schemas and the number of non-joinable fields.

• AS1. The dataset is inherited from the Data Collection and Preparation, Data Analysis with Python, and Statistical Analysis and Exploratory Data Analysis courses and is fictional: 3,199 rows, 1,329 subscribers, three periods, five regions, tariff tiers of 10/25/40 m³. It is generated with the same seed and the output is the same from run to run. It is a model.
• AS2. Bytes per row are measured with a single representation. Which decision determines the bytes that the field types set apart was measured in the Scale and Performance topic of the Data Analysis with Python course; here that decision is an input and is taken as a single number.
• AS3. The working space of a single machine is an assumption and is tested in the table with three values. The intermediate share is the second decision: whether temporary space as large as the dataset itself is requested next to the dataset.
• AS4. Velocity is two numbers: rows/period and latency. Latency is the largest number of reading intervals a reading waits in a report that is refreshed periodically.
• AS5. Variety is two numbers: the number of distinct schemas and the number of non-joinable fields. A field does not join if it is in a schema with no common key, if it is free text with no mapping, if it is a quantity whose unit must be converted, or if its time format does not fit the dataset’s period axis.
• AS6. No distributed processing engine, file format, or vendor name is written; tools are mentioned by their kind.

Volume: Bytes and Rows

# dataset.py — the FICTIONAL dataset of the regional measurement network. MODEL: instead of
# a real dataset a generator that produces the same structure was written; the seed is visible and the
# output is the same from run to run. 3199 rows, 1329 subscribers, three periods, five regions.
import math, sys

SEED, RAW, M32 = 20260218, 1400, 0xFFFFFFFF
REGION = [("north", 0.28, 21), ("south", 0.22, 17), ("east", 0.18, 26),
          ("west", 0.14, 14), ("center", 0.18, 23)]
FIELDS = ("subscriber_id", "region", "period", "household", "satisfaction",
          "consumption_m3")


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


DATASET = []
for i in range(RAW):
    r = rng(SEED + i)
    b = REGION[pick(r(), [x[1] for x in REGION])]
    household = 1 + pick(r(), [0.06, 0.24, 0.30, 0.24, 0.11, 0.05])
    satisfied = 1 + pick(r(), [0.08, 0.14, 0.27, 0.34, 0.17])
    if r() < 0.046:                      # meter that failed to register
        continue
    sub_id, q = 10001 + i, rng(SEED + 7000 + 10001 + i)
    for d in range(pick(q(), [0.05, 0.12, 0.21, 0.62])):
        v = 0.0 if q() < 0.038 else math.floor(
            b[2] * math.exp((q() + q() + q() - 1.5) * 0.62)
            * (1 - d * 0.05) * 100 + 0.5) / 100
        DATASET.append((sub_id, b[0], d + 1, household, satisfied, v))
        q()

METERS = len({s[0] for s in DATASET})
total_bytes = sys.getsizeof(DATASET) + sum(
    sys.getsizeof(s) + sum(sys.getsizeof(a) for a in s) for s in DATASET)
BYTES_PER_ROW = total_bytes / len(DATASET)
print(f"seed {SEED}; {len(DATASET)} rows, {len(FIELDS)} fields, {METERS} reading meters")
print(f"in memory {total_bytes} bytes, {BYTES_PER_ROW:.1f} bytes per row")

GIB = 1024 ** 3
print(f"{'working space':<16}{'intermediate share':>20}{'left for dataset':>18}"
      f"{'threshold rows':>16}{'times':>8}")
for gib in (4, 12, 64):
    for share, label in ((1, "none"), (2, "same as dataset")):
        left = gib * GIB / share
        threshold = int(left / BYTES_PER_ROW)
        print(f"{str(gib) + ' GiB':<16}{label:>20}{gib // share:>14} GiB"
              f"{threshold:>16}{threshold // len(DATASET):>8}")

seed 20260218; 3199 rows, 6 fields, 1260 reading meters
in memory 914929 bytes, 286.0 bytes per row
working space     intermediate share  left for dataset  threshold rows   times
4 GiB                           none             4 GiB        15017121    4694
4 GiB                same as dataset             2 GiB         7508560    2347
12 GiB                          none            12 GiB        45051365   14082
12 GiB               same as dataset             6 GiB        22525682    7041
64 GiB                          none            64 GiB       240273951   75109
64 GiB               same as dataset            32 GiB       120136975   37554

Volume is not a single number but the quotient of two numbers. The working space stands on the numerator side, bytes per row on the denominator side, and the quotient gives a row count. The fictional dataset takes 914,929 bytes, 286.0 bytes per row. This number is itself the result of a type decision, and that decision was measured elsewhere; what does not change here is that the threshold can be written as a number.

The six rows of the table are the product of two defensible decisions. The working space takes three values, the intermediate share two. The narrowest decision reaches the threshold at 7,508,560 rows, the widest at 240,273,951 rows. The ratio between them is thirty-two times and both ends are defensible: a computation that asks for no intermediate share keeps only running totals, while a computation that asks for temporary space as large as the dataset itself sorts or merges. The sentence “this dataset does not fit on a single machine” cannot be formed without writing which of the six rows was chosen.

The fictional dataset itself comes close to the threshold in no row: even under the narrowest decision it is one 2,347th of the threshold. The subject of analysis at scale is not this dataset but the speed at which this dataset walks toward that threshold.

The next two sections take the middle row as the threshold, 12 GiB of working space with an intermediate share the size of the dataset, and fix 22,525,682 rows. This is a choice and it is written down because it is a choice; when the threshold changes, all the later numbers shift by the same ratio and the direction of the shift can be read from the table.

Velocity: Rows and Periods

# velocity.py — velocity is two numbers: rows/period and latency. Reading frequency is a
# decision; the meter itself does not change, yet the number of rows it produces does.
THRESHOLD = int((12 * GIB / 2) / BYTES_PER_ROW)     # 12 GiB working space, intermediate share the size of the dataset
GROWTH = 1.04                                       # meter growth per period


def period_count(rate, growth):
    total, n, h = len(DATASET), 0, rate
    while total < THRESHOLD and n < 10 ** 6:
        total, n, h = total + h, n + 1, h * growth
    return n


print(f"threshold {THRESHOLD} rows; network {METERS} meters, growth per period "
      f"{round((GROWTH - 1) * 100)} percent")
print(f"{'reading frequency':<20}{'reads/period':>14}{'rows/period':>13}"
      f"{'fixed network':>15}{'growing network':>17}{'latency':>9}")
for label, freq in (("periodic", 1), ("daily", 30), ("hourly", 720),
                    ("fifteen-minute", 2880)):
    rate = METERS * freq
    print(f"{label:<20}{freq:>14}{rate:>13}{period_count(rate, 1.0):>15}"
          f"{period_count(rate, GROWTH):>17}{freq:>9}")

threshold 22525682 rows; network 1260 meters, growth per period 4 percent
reading frequency     reads/period  rows/period  fixed network  growing network  latency
periodic                         1         1260          17875              168        1
daily                           30        37800            596               82       30
hourly                         720       907200             25               18      720
fifteen-minute                2880      3628800              7                6     2880

Velocity is not the growth of the network. The four rows of the table hold the same 1,260 meters; the only thing that changes is how many times each meter is read. Under periodic reading the network produces 1,260 rows/period and the threshold is 17,875 periods away; under fifteen-minute reading the same network produces 3,628,800 rows/period and the threshold arrives 7 periods later. Reading frequency looks like a hardware fact, yet it is a decision, and the effect of this decision on the periods left to the threshold is 2,554 times.

The fifth column adds the second decision. When the network grows 4% per period, the 17,875 periods under periodic reading drop to 168 periods. A growth of 4% is invisible when looked at period by period; looked at cumulatively, it brings the threshold one hundred times closer. Velocity has two sources and the two are multiplied: frequency and growth.

The last column is the second unit of velocity. If a report is refreshed once per period, it takes a reading at most one period to enter the report; but when this wait is written in terms of reading intervals, the number changes. Under periodic reading the wait is 1 reading interval, under fifteen-minute reading it is 2,880 reading intervals, and the rows waiting outside the report in the meantime rise from 1,260 to 3,628,800. Raising the frequency does not freshen the data; it enlarges the amount of data that has not been refreshed.

Variety: Schema and Field

# variety.py — variety is two numbers: the number of distinct schemas and the number of
# non-joinable fields. MODEL: the schemas of five sources were written as field name, unit, and
# time format; the sources themselves were not generated.
COMMON = "subscriber_id"
SCHEMA = {
    "subscriber record": [("subscriber_id", "-", "-"), ("region", "name", "-"),
                          ("household", "persons", "-"), ("satisfaction", "1-5", "-")],
    "meter reading": [("subscriber_id", "-", "-"), ("period", "-", "period no"),
                      ("consumption_m3", "m3", "-")],
    "field visit": [("meter_serial", "-", "-"), ("value_read", "liter", "-"),
                    ("region_code", "number", "-"), ("visit_day", "-", "day")],
    "support request": [("request_id", "-", "-"), ("subscriber_id", "-", "-"),
                        ("opened", "-", "timestamp"), ("topic", "text", "-")],
    "tariff table": [("tier_text", "text", "-"), ("unit_price", "currency", "-"),
                     ("valid_range", "-", "date range")],
}


def status(schema, unit, time_format):
    if COMMON not in [a for a, _, _ in SCHEMA[schema]]:
        return "no key"
    if unit == "liter":
        return "unit conversion"
    if unit == "text":
        return "unmapped text"
    if time_format not in ("-", "period no"):
        return "time format"
    return "joins"


def count(title):
    reasons, broken = {}, 0
    print(f"{title:<18}{'fields':>8}{'common key':>15}{'non-joinable':>14}"
          f"{'reason':>18}")
    for name, fields in SCHEMA.items():
        d = [status(name, u, t) for _, u, t in fields]
        n = sum(1 for x in d if x != "joins")
        for x in d:
            if x != "joins":
                reasons[x] = reasons.get(x, 0) + 1
        broken += n
        print(f"{name:<18}{len(fields):>8}"
              f"{('yes' if COMMON in [a for a, _, _ in fields] else 'no'):>15}"
              f"{n:>14}{(sorted(set(x for x in d if x != 'joins')) or ['-'])[0]:>18}")
    print(f"distinct schemas {len(SCHEMA)}, total fields "
          f"{sum(len(v) for v in SCHEMA.values())}, non-joinable fields {broken}")
    print("reason breakdown:", ", ".join(f"{k} {v}" for k, v in sorted(reasons.items())))
    return broken


k1 = count("schema (before)")
SCHEMA["field visit"].append(("subscriber_id", "-", "-"))     # meter_serial mapping table
SCHEMA["tariff table"].append(("subscriber_id", "-", "-"))    # tier mapping table
print()
k2 = count("schema (after)")
print(f"two mapping tables reduce non-joinable fields from {k1} to {k2}; "
      f"key absence closes, unit and time format do not close")

schema (before)     fields     common key  non-joinable            reason
subscriber record        4            yes             0                 -
meter reading            3            yes             0                 -
field visit              4             no             4            no key
support request          4            yes             2       time format
tariff table             3             no             3            no key
distinct schemas 5, total fields 18, non-joinable fields 9
reason breakdown: no key 7, time format 1, unmapped text 1

schema (after)      fields     common key  non-joinable            reason
subscriber record        4            yes             0                 -
meter reading            3            yes             0                 -
field visit              5            yes             2       time format
support request          4            yes             2       time format
tariff table             4            yes             2       time format
distinct schemas 5, total fields 20, non-joinable fields 6
reason breakdown: time format 3, unit conversion 1, unmapped text 2
two mapping tables reduce non-joinable fields from 9 to 6; key absence closes, unit and time format do not close

The number of distinct schemas does not measure variety. Five sources carry five schemas and this number stays five in both runs; the number that moves is the number of non-joinable fields: 9 of the 18 fields cannot be tied to the dataset. The reason breakdown splits these nine into three — 7 fields come from two schemas that have no common key, 1 field is free text with no mapping, and 1 field carries a time format that does not fit the dataset’s period axis.

The second run applies a decision: a mapping table that converts the meter serial number to the subscriber number is built, along with a second mapping table that converts the tariff tier text to the dataset’s tier names. Non-joinable fields fall from 9 to 6, but the reason breakdown shifts: key absence closes completely, while time format conflicts rise from 1 to 3, because the time fields of the two schemas that can now be tied are counted as well. A mapping ties a field; it does not align its unit and its time axis.

The direction of the number is also worth noting. In the second run the total number of fields rises from 18 to 20, because the mapping adds one field to each of two schemas; the share of non-joinable fields falls from 50% to 30%, but the absolute number falls by only three. The cost of closing variety is a mapping table for every field that closes, one that must keep being maintained, and these tables do not appear in the dataset itself.

When the three units are set side by side, the scale table of the dataset emerges. The volume threshold is 22,525,682 rows, the velocity is 3,628,800 rows/period under fifteen-minute reading, and the variety is 5 schemas and 6 non-joinable fields. None of these three numbers is “big” or “small”; each is the quotient of a pair of decisions, and the decisions are written inside the table.

Summary

• Volume is the quotient of two numbers: the fictional dataset takes 286.0 bytes per row, and the threshold at which it no longer fits on a single machine moves thirty-two times, between 7,508,560 and 240,273,951 rows, depending on the working space and intermediate share decisions.
• Velocity is the reading frequency, not the growth of the network: the same 1,260 meters produce 1,260 rows/period under periodic reading and 3,628,800 rows/period under fifteen-minute reading, and the periods left to the threshold fall from 17,875 to 7.
• A network growth of 4% per period brings the threshold closer under periodic reading from 17,875 periods to 168 periods; frequency and growth are multiplied.
• Latency is the second unit of velocity: in a report refreshed periodically the wait is 1 reading interval under periodic reading and 2,880 reading intervals under fifteen-minute reading, and the waiting rows rise to 3,628,800.
• The number of distinct schemas does not measure variety; five schemas are five in both runs, the number that moves is the 9 non-joinable fields out of 18, and two mapping tables bring this number down to 6.
• A mapping closes key absence but does not close the unit and time format conflict: the number of fields that do not join because of the time format rises from 1 to 3.

Next Step

This lesson wrote the point at which the dataset no longer fits on a single machine as a row count, but it did not ask what the computation becomes after that point. When the dataset crosses the threshold, the work has to be divided, and the first question of dividing is this: into which pieces can an aggregation be split, and how many records are moved between the pieces. The next lesson writes and runs the simplest abstraction of distributed computation, the map and reduce functions, within the lesson; the measure is the number of records moved in the shuffle — how many records move when the same aggregation first reduces and then shuffles and when it does the reverse, and which computation does not enter this model at all.