### IOT Data Example Setup

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/repeatable_data_generation.rst

Sets up Spark configuration, defines data parameters, and lists possible values for country codes, manufacturers, and phone lines for simulating IOT events.

```python
from pyspark.sql.types import LongType, IntegerType, StringType

import dbldatagen as dg

shuffle_partitions_requested = 8
device_population = 100000
data_rows = 20 * 1000000
partitions_requested = 20

spark.conf.set("spark.sql.shuffle.partitions", shuffle_partitions_requested)

country_codes = ['CN', 'US', 'FR', 'CA', 'IN', 'JM', 'IE', 'PK', 'GB', 'IL', 'AU', 'SG',
                 'ES', 'GE', 'MX', 'ET', 'SA', 'LB', 'NL']
country_weights = [1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 126, 109, 58, 8,
                   17]

manufacturers = ['Delta corp', 'Xyzzy Inc.', 'Lakehouse Ltd', 'Acme Corp', 'Embanks Devices']

lines = ['delta', 'xyzzy', 'lakehouse', 'gadget', 'droid']

testDataSpec = (

```

--------------------------------

### Using SQL expressions in data generation

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

Example demonstrating the use of SQL expressions for generating synthetic names, email addresses, and computing MD5 hashes of payment instruments.

```python
import dbldatagen as dg

shuffle_partitions_requested = 8
partitions_requested = 8
data_rows = 10000000

spark.conf.set("spark.sql.shuffle.partitions", shuffle_partitions_requested)

dataspec = (
    dg.DataGenerator(spark, rows=data_rows, partitions=8)
    .withColumn("name", percentNulls=0.01, template=r'\\w \\w|\\w a. \\w') 
    .withColumn("payment_instrument_type", values=['paypal', 'visa', 'mastercard', 'amex'], 
                random=True)             
    .withColumn("int_payment_instrument", "int",  minValue=0000, maxValue=9999,  
                baseColumn="name",
                baseColumnType="hash", omit=True)
    .withColumn("payment_instrument", 
                 expr="format_number(int_payment_instrument, '**** ****** *####')",
                 baseColumn="int_payment_instrument")
    .withColumn("email", template=r'\\w.\\w@\\w.com')       
    .withColumn("md5_payment_instrument", 
                expr="md5(concat(payment_instrument_type, ':', payment_instrument))",
                baseColumn=['payment_instrument_type', 'payment_instrument']) 
   )
df1 = dataspec.build()

df1.show()
```

--------------------------------

### Generating IOT Device Style Data

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

This example demonstrates the generation of IOT device style data, ensuring that device properties remain consistent across events for the same device.

```python
from pyspark.sql.types import LongType, IntegerType, StringType

import dbldatagen as dg

shuffle_partitions_requested = 8
device_population = 100000
data_rows = 20 * 1000000
partitions_requested = 20

spark.conf.set("spark.sql.shuffle.partitions", shuffle_partitions_requested)

country_codes = [
    "CN", "US", "FR", "CA", "IN", "JM", "IE", "PK", "GB", "IL", "AU", 
    "SG", "ES", "GE", "MX", "ET", "SA", "LB", "NL",
]
country_weights = [
    1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 
    126, 109, 58, 8, 17,
]

manufacturers = [
    "Delta corp", "Xyzzy Inc.", "Lakehouse Ltd", "Acme Corp", "Embanks Devices",
]

lines = ["delta", "xyzzy", "lakehouse", "gadget", "droid"]

testDataSpec = (
    dg.DataGenerator(spark, name="device_data_set", rows=data_rows, 
                     partitions=partitions_requested)
    .withIdOutput()
    # we'll use hash of the base field to generate the ids to
    # avoid a simple incrementing sequence
    .withColumn("internal_device_id", "long", minValue=0x1000000000000, 
                uniqueValues=device_population, omit=True, baseColumnType="hash",
    )
    # note for format strings, we must use "%lx" not "%x" as the
    # underlying value is a long
    .withColumn(
        "device_id", "string", format="0x%013x", baseColumn="internal_device_id"
    )
    # the device / user attributes will be the same for the same device id
    # so lets use the internal device id as the base column for these attribute
    .withColumn("country", "string", values=country_codes, weights=country_weights, 
                baseColumn="internal_device_id")
    .withColumn("manufacturer", "string", values=manufacturers, 
                baseColumn="internal_device_id", )
    # use omit = True if you don't want a column to appear in the final output
    # but just want to use it as part of generation of another column
    .withColumn("line", "string", values=lines, baseColumn="manufacturer", 
                baseColumnType="hash", omit=True )
    .withColumn("model_ser", "integer", minValue=1, maxValue=11, baseColumn="device_id", 
                baseColumnType="hash", omit=True, )
    .withColumn("model_line", "string", expr="concat(line, '#', model_ser)", 
                baseColumn=["line", "model_ser"] )
    .withColumn("event_type", "string", 
                values=["activation", "deactivation", "plan change", "telecoms activity", 
                        "internet activity", "device error", ],
                random=True)
    .withColumn("event_ts", "timestamp", begin="2020-01-01 01:00:00", 
                end="2020-12-31 23:59:00", 
                interval="1 minute", random=True )
)

dfTestData = testDataSpec.build()

display(dfTestData)

```

--------------------------------

### Install dbldatagen from PyPi using pip (command line)

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs the dbldatagen package from PyPi into a command-line environment.

```bash
pip install dbldatagen
```

--------------------------------

### Install an older release from PyPi using version qualifier

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs a specific older release of dbldatagen from PyPi by specifying the version.

```python
%pip install dbldatagen==v0.3.0
```

--------------------------------

### Install a specific Python wheel release using %pip

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs a specific dbldatagen Python wheel file directly from a GitHub release asset.

```python
%pip install https://github.com/databrickslabs/dbldatagen/releases/download/v021/dbldatagen-0.2.1-py3-none-any.whl
```

--------------------------------

### Install dbldatagen from PyPi using %pip

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs the dbldatagen package from PyPi into a Databricks notebook environment.

```python
%pip install dbldatagen
```

--------------------------------

### Generating repeatable data with a random seed

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

Example showing how to generate repeatable random data across runs by specifying a random seed method and value.

```python
import dbldatagen as dg

shuffle_partitions_requested = 8
partitions_requested = 8
data_rows = 10000000

spark.conf.set("spark.sql.shuffle.partitions", shuffle_partitions_requested)

dataspec = (
    dg.DataGenerator(spark, rows=data_rows, partitions=8, randomSeedMethod="hash_fieldname", 
                     randomSeed=42)
    .withColumn("name", percentNulls=0.01, template=r'\\w \\w|\\w a. \\w')
    .withColumn("payment_instrument_type", values=['paypal', 'visa', 'mastercard', 'amex'],
                random=True)
    .withColumn("int_payment_instrument", "int",  minValue=0000, maxValue=9999,  
                baseColumn="name",
                baseColumnType="hash", omit=True)
    .withColumn("payment_instrument", 
                expr="format_number(int_payment_instrument, '**** ****** *####')",
                baseColumn="int_payment_instrument")
    .withColumn("email", template=r'\\w.\\w@\\w.com')
    .withColumn("md5_payment_instrument",
                expr="md5(concat(payment_instrument_type, ':', payment_instrument))",
                baseColumn=['payment_instrument_type', 'payment_instrument'])
    )
df1 = dataspec.build()

df1.show()
```

--------------------------------

### Create a data set without pre-existing schemas

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

Example of creating a simple synthetic data set without the use of a schema.

```python
import dbldatagen as dg
from pyspark.sql.types import FloatType, IntegerType, StringType

row_count = 1000 * 100
column_count = 10
testDataSpec = (
    dg.DataGenerator(spark, name="test_data_set1", rows=row_count, partitions=4)
    .withIdOutput()
    .withColumn(
        "r",
        FloatType(),
        expr="floor(rand() * 350) * (86400 + 3600)",
        numColumns=column_count,
    )
    .withColumn("code1", IntegerType(), minValue=100, maxValue=200)
    .withColumn("code2", "integer", minValue=0, maxValue=10, random=True)
    .withColumn("code3", StringType(), values=["online", "offline", "unknown"])
    .withColumn(
        "code4", StringType(), values=["a", "b", "c"], random=True, percentNulls=0.05
    )
    .withColumn(
        "code5", "string", values=["a", "b", "c"], random=True, weights=[9, 1, 1]
    )
)

dfTestData = testDataSpec.build()
```

--------------------------------

### Install dbldatagen from Databricks Labs Github repository using %pip

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs and builds the dbldatagen package from the latest release on the master branch of the Databricks Labs Github repository.

```python
%pip install git+https://github.com/databrickslabs/dbldatagen@current
```

--------------------------------

### Install dbldatagen from a specific branch/tag on Github

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/installation_notes.rst

Installs dbldatagen from a specific branch or tag on the Databricks Labs Github repository.

```python
%pip install git+https://github.com/databrickslabs/dbldatagen@dbr_7_3_LTS_compat
```

--------------------------------

### Import dbldatagen framework

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

Imports the dbldatagen framework in Python code for use.

```python
import dbldatagen as dg
```

--------------------------------

### Install Faker in Databricks Notebook

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/extending_text_generation.rst

Installs the Faker library using the %pip magic command.

```python
%pip install Faker
```

--------------------------------

### Ipsum Lorem Text Generator Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/textdata.rst

Example demonstrating the use of the Ipsum Lorem text generator to create text columns with specified paragraph and sentence counts.

```python
import dbldatagen as dg
df_spec = (
   dg.DataGenerator(sparkSession=spark, name="test_data_set1", rows=100000,
                     partitions=4, randomSeedMethod="hash_fieldname")
   .withIdOutput()
   .withColumn("sample_text", "string", text=dg.ILText(paragraphs=(1, 4),
                   sentences=(2, 6)))
)

df = df_spec.build()
num_rows=df.count()
```

--------------------------------

### Random Data Generation Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/options_and_features.rst

Illustrates features like random data generation, array-valued columns, and templates.

```python
ds = (
            dg.DataGenerator(sparkSession=spark, name="test_dataset1", rows=1000, partitions=4,
                             random=True)
            .withColumn("name", "string", percentNulls=0.01, template=r'\\w \\w|\\w A. \\w|test')
            .withColumn("emails", "string", template=r'\\w.\\w@\\w.com', random=True,
                        numFeatures=numFeaturesSupplied, structType="array")
        )

        df = ds.build()
```

--------------------------------

### Example 2: IOT style data

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_streaming_data.rst

This example demonstrates generating IoT-style streaming data, focusing on controlling the duration of the data generation. It sets a `time_to_run` variable and uses various column specifications for device attributes like country, manufacturer, and device ID.

```python
import time
time_to_run = 180

from pyspark.sql.types import LongType, IntegerType, StringType

import dbldatagen as dg

device_population = 10000
data_rows = 20 * 100000
partitions_requested = 8

country_codes = ['CN', 'US', 'FR', 'CA', 'IN', 'JM', 'IE', 'PK', 'GB', 'IL', 'AU', 'SG',
                 'ES', 'GE', 'MX', 'ET', 'SA', 'LB', 'NL']
country_weights = [1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 126, 109, 58,
                   8, 17]

manufacturers = ['Delta corp', 'Xyzzy Inc.', 'Lakehouse Ltd', 'Acme Corp', 'Embanks Devices']

lines = ['delta', 'xyzzy', 'lakehouse', 'gadget', 'droid']

testDataSpec = (
    dg.DataGenerator(spark, name="device_data_set", rows=data_rows,
                     partitions=partitions_requested,
                     verbose=True)
    .withIdOutput()
    # we'll use hash of the base field to generate the ids to
    # avoid a simple incrementing sequence
    .withColumn("internal_device_id", LongType(), minValue=0x1000000000000,
                uniqueValues=device_population, omit=True, baseColumnType="hash")

    # note for format strings, we must use "%lx" not "%x" as the
    # underlying value is a long
    .withColumn("device_id", StringType(), format="0x%013x",
                baseColumn="internal_device_id")

    # the device / user attributes will be the same for the same device id
    # so lets use the internal device id as the base column for these attribute
    .withColumn("country", StringType(), values=country_codes,
                weights=country_weights, baseColumn="internal_device_id")
    .withColumn("manufacturer", StringType(), values=manufacturers,
                baseColumn="internal_device_id")

    # use omit = True if you don't want a column to appear in the final output
    # but just want to use it as part of generation of another column
    .withColumn("line", StringType(), values=lines, baseColumn="manufacturer",
                baseColumnType="hash", omit=True)
)

```

--------------------------------

### Example of using constraints

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_constraints.rst

This example demonstrates how to use a SQL constraint to ensure that the shipping timestamp is not earlier than the order timestamp.

```python
import dbldatagen as dg

data_rows = 10000000

dataspec = dg.DataGenerator(spark, rows=10000000, partitions=8)

dataspec = (
    dataspec.withColumn("name", "string", template=r"\\w \\w|\\w a. \\w")
    .withColumn(
        "product_sku", "string", minValue=1000000, maxValue=1000000 + 1000, prefix="dr", random=True
    )
    .withColumn("email", "string", template=r"\\w.\\w@\\w.com")
    .withColumn("qty_ordered", "int", minValue=1, maxValue=10, distribution="normal", random=True)
    .withColumn("unit_price", "float", minValue=1.0, maxValue=30.0, step=0.01, distribution="normal",
                baseColumn="product_sku", baseColumnType="hash")
    .withColumn("order_ts", "timestamp", begin="2020-01-01 01:00:00",
                end="2020-12-31 23:59:00",
                interval="1 minute", random=True )
    .withColumn("shipping_ts", "timestamp", begin="2020-01-05 01:00:00",
                end="2020-12-31 23:59:00",
                interval="1 minute", random=True, percentNulls=0.1)
    .withSqlConstraint(""shipping_ts is null or shipping_ts > order_ts"")
)
df1 = dataspec.build()
```

--------------------------------

### Streaming Data Generation Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_streaming_data.rst

Example of building a streaming DataFrame with specified columns and options.

```python
testDataSpec = (
       dg.StreamingSpec(spark, rowsPerSecond=500)
       .withColumn("model_ser", IntegerType(), minValue=1, maxValue=11,
                   baseColumn="device_id", baseColumnType="hash", omit=True)
       .withColumn("model_line", StringType(), expr="concat(line, '#', model_ser)",
                   baseColumn=["line", "model_ser"])
       .withColumn("event_type", StringType(),
                   values=["activation", "deactivation", "plan change",
                           "telecoms activity", "internet activity", "device error"],
                   random=True)
       .withColumn("event_ts", "timestamp", expr="now()")
       )

dfTestDataStreaming = testDataSpec.build(withStreaming=True, options={'rowsPerSecond': 500})
```

--------------------------------

### DataGenerator Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/repeatable_data_generation.rst

Example of using DataGenerator to create a device data set with various columns and configurations.

```python
dg.DataGenerator(spark, name="device_data_set", rows=data_rows,
                         partitions=partitions_requested,
                         randomSeedMethod='hash_fieldname')
       .withIdOutput()
       # we'll use hash of the base field to generate the ids to
       # avoid a simple incrementing sequence
       .withColumn("internal_device_id", LongType(), minValue=0x1000000000000,
                   uniqueValues=device_population, omit=True, baseColumnType="hash")

       # note for format strings, we must use "%lx" not "%x" as the
       # underlying value is a long
       .withColumn("device_id", StringType(), format="0x%013x",
                   baseColumn="internal_device_id")

       # the device / user attributes will be the same for the same device id
       # so lets use the internal device id as the base column for these attribute
       .withColumn("country", StringType(), values=country_codes,
                   weights=country_weights,
                   baseColumn="internal_device_id")
       .withColumn("manufacturer", StringType(), values=manufacturers,
                   baseColumn="internal_device_id")

       # use omit = True if you don't want a column to appear in the final output
       # but just want to use it as part of generation of another column
       .withColumn("line", StringType(), values=lines,
                   baseColumn="manufacturer",
                   baseColumnType="hash", omit=True)
       .withColumn("model_ser", IntegerType(), minValue=1, maxValue=11,
                   baseColumn="device_id",
                   baseColumnType="hash", omit=True)

       .withColumn("model_line", StringType(), expr="concat(line, '#', model_ser)",
                   baseColumn=["line", "model_ser"])
       .withColumn("event_type", StringType(),
                   values=["activation", "deactivation", "plan change",
                           "telecoms activity", "internet activity", "device error"],
                   random=True)
       .withColumn("event_ts", "timestamp",
                   begin="2020-01-01 01:00:00", end="2020-12-31 23:59:00",
                   interval="1 minute",
                   random=True)

       )

   dfTestData = testDataSpec.build()

   display(dfTestData)
```

--------------------------------

### Simple Date Range Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/DATARANGES.md

This example demonstrates specifying a simple date range by setting the number of unique dates.

```python
import dbldatagen as dg
from pyspark.sql.types import IntegerType

row_count=1000 * 100
testDataSpec = (
      dg.DataGenerator(spark, name="test_data_set1", rows=row_count,
                       partitions=4, randomSeedMethod='hash_fieldname',
                       verbose=True)
      .withColumn("purchase_id", IntegerType(), minValue=1000000, 
                     maxValue=2000000)
      .withColumn("product_code", IntegerType(), uniqueValues=10000, 
                     random=True)
      .withColumn("purchase_date", "date", uniqueValues=300, 
                     random=True)
      )

dfTestData = testDataSpec.build()
```

--------------------------------

### Adding multiple column specifications using patterns and types

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

This example shows how to use `withColumnSpecs` to apply generation rules to multiple columns based on naming patterns and data types. It demonstrates overriding specific column rules with `withColumnSpec`.

```python
(dg
   .withColumnSpecs(patterns=".*_ID", matchTypes=StringType(), format="%010d", 
                           minValue=10, maxValue=123, step=1)
   .withColumnSpecs(patterns=".*_IDS", matchTypes=StringType(), format="%010d", 
                           minValue=1, maxValue=100, step=1)
)
```

--------------------------------

### Build and serve documentation locally

Source: https://github.com/databrickslabs/dbldatagen/blob/master/CONTRIBUTING.md

Builds the HTML documentation and serves it locally for review.

```bash
make docs-serve
```

--------------------------------

### Generating data generation code from a Spark DataFrame

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

This snippet illustrates how to use the `DataAnalyzer` class to generate synthetic data generation code from an existing Spark DataFrame. This is useful for quick prototyping.

```python
import dbldatagen as dg

dfSource = spark.read.format("parquet").load("/tmp/your/source/dataset")

analyzer = dg.DataAnalyzer(sparkSession=spark, df=df_source_data)

generatedCode = analyzer.scriptDataGeneratorFromData()
```

--------------------------------

### Generating a Billion Rows of Data

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

This code snippet demonstrates how to configure dbldatagen to generate one billion rows of data, specifying the number of rows and Spark partitions, and then writing the data to a Delta table.

```python
from pyspark.sql.types import LongType, IntegerType, StringType

import dbldatagen as dg

shuffle_partitions_requested = 12 * 4
partitions_requested = 96
device_population = 100000
data_rows = 1000 * 1000000

spark.conf.set("spark.sql.shuffle.partitions", shuffle_partitions_requested)

country_codes = ['CN', 'US', 'FR', 'CA', 'IN', 'JM', 'IE', 'PK', 'GB', 'IL', 'AU', 'SG', 'ES',
                 'GE', 'MX', 'ET', 'SA', 'LB', 'NL']
country_weights = [1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 126, 109, 58, 8,
                   17]

manufacturers = ['Delta corp', 'Xyzzy Inc.', 'Lakehouse Ltd', 'Acme Corp', 'Embanks Devices']

lines = ['delta', 'xyzzy', 'lakehouse', 'gadget', 'droid']

testDataSpec = (
    dg.DataGenerator(spark, name="device_data_set", rows=data_rows,
                                 partitions=partitions_requested)
    .withIdOutput()
    # we'll use hash of the base field to generate the ids to 
    # avoid a simple incrementing sequence
    .withColumn("internal_device_id", LongType(), minValue=0x1000000000000,
                uniqueValues=device_population, omit=True, baseColumnType="hash")

    # note for format strings, we must use "%lx" not "%x" as the 
    # underlying value is a long
    .withColumn("device_id", StringType(), format="0x%013x",
                baseColumn="internal_device_id")

    # the device / user attributes will be the same for the same device id 
    # so lets use the internal device id as the base column for these attribute
    .withColumn("country", StringType(), values=country_codes,
                weights=country_weights,
                baseColumn="internal_device_id")
    .withColumn("manufacturer", StringType(), values=manufacturers,
                baseColumn="internal_device_id")

    # use omit = True if you don't want a column to appear in the final output 
    # but just want to use it as part of generation of another column
    .withColumn("line", StringType(), values=lines, baseColumn="manufacturer",
                baseColumnType="hash", omit=True)
    .withColumn("model_ser", IntegerType(), minValue=1, maxValue=11,
                baseColumn="device_id",
                baseColumnType="hash", omit=True)

    .withColumn("model_line", StringType(), expr="concat(line, '#', model_ser)",
                baseColumn=["line", "model_ser"])
    .withColumn("event_type", StringType(),
                values=["activation", "deactivation", "plan change",
                        "telecoms activity", "internet activity", "device error"],
                random=True)
    .withColumn("event_ts", "timestamp", begin="2020-01-01 01:00:00", 
                end="2020-12-31 23:59:00",
                interval="1 minute", 
                random=True)

    )

dfTestData = testDataSpec.build()

dfTestData.write.format("delta").mode("overwrite").save(
    "/tmp/dbldatagen_examples/a_billion_row_table")
```

--------------------------------

### Generating data with a pre-existing table schema

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/APIDOCS.md

This snippet demonstrates how to create a data generator that uses the schema of an existing Spark SQL table. It defines column specifications for name, serial_number, email, and license_plate, and then builds and saves the generated data.

```python
import dbldatagen as dg
from pyspark.sql.types import StructType, StructField,  StringType

shuffle_partitions_requested = 8
partitions_requested = 8
data_rows = 10000000

spark.sql("""Create table if not exists test_vehicle_data(
                name string, 
                serial_number string, 
                license_plate string, 
                email string
                ) using Delta""")

table_schema = spark.table("test_vehicle_data").schema

print(table_schema)
  
dataspec = (dg.DataGenerator(spark, rows=10000000, partitions=8)
            .withSchema(table_schema))

dataspec = (
    dataspec.withColumnSpec("name", percentNulls=0.01, template=r"\\w \\w|\\w a. \\w")
    .withColumnSpec(
        "serial_number", minValue=1000000, maxValue=10000000, prefix="dr", random=True
    )
    .withColumnSpec("email", template=r"\\w.\\w@\\w.com")
    .withColumnSpec("license_plate", template=r"\\n-\\n")
)
df1 = dataspec.build()

df1.write.format("delta").mode("overwrite").saveAsTable("test_vehicle_data")
```

--------------------------------

### Creating data generators from configuration

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/serialized_data_generators.rst

Illustrates how to create a DataGenerator object from a Python dictionary using the loadFromInitializationDict method.

```python
import dbldatagen as dg

# Define the data generation options:
dataSpecOptions = {
   "name": "users_dataset",
   "rows": 1000,
   "randomSeedMethod": "hash_fieldname",
   "columns": [
      {"colName": "user_name", "colType": "string", "expr": "concat('user_', id)"},
      {"colName": "phone_number", "colType": "string", "template": "555-DDD-DDDD"}
   ]
}

# Create the DataGenerator from options:
dg.DataGenerator.loadFromInitializationDict(dataSpecOptions)
```

--------------------------------

### Device Events Data Generation Setup

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/multi_table_data.rst

Initializes Spark configuration and imports necessary libraries for generating device events, setting up for master-detail style data modeling.

```python
import dbldatagen as dg
import pyspark.sql.functions as F

AVG_EVENTS_PER_CUSTOMER = 50

spark.catalog.clearCache()
```

--------------------------------

### Custom SQL and Array Generation Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/options_and_features.rst

Shows the use of custom SQL expressions, controlling column order, and generating user records with variable email addresses.

```python
import dbldatagen as dg
import logging

from pyspark.sql.types import ArrayType, StringType

dataspec = dg.DataGenerator(spark, rows=10 * 1000000)

logging.info(dataspec.partitions)

dataspec = (
         dataspec
         .withColumn("name", "string", percentNulls=0.01, template=r'\\w \\w|\\w A. \\w|test')
         .withColumn("serial_number", "string",
                     minValue=1000000, maxValue=10000000,
                     prefix="dr", random=True)

        # generate a fixed length array of email addresses
        .withColumn("email", "string", template=r'\\w.\\w@\\w.com', omit=True,
                    numColumns=5, structType="array",
                    random=True, randomSeed=-1)
        .withColumn("emailCount", "int", expr="abs(hash(id)) % 4)+1)")
        .withColumn("emails", ArrayType(StringType()), expr="slice(email, 1, emailCount",
                        baseColumns=["email"])
         .withColumn("license_plate", "string", template=r'\\n-\\n')
        )
dfTestData = dataspec.build()

display(dfTestData)
```

--------------------------------

### Multi-table Dataset Generation and Joining

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_standard_datasets.rst

This code snippet shows how to generate multiple related datasets (plans, customers, deviceEvents, invoices) using the multi-table provider and then join them, which is useful for benchmarking.

```python
import dbldatagen as dg

multiTableDS = dg.Datasets(spark, "multi_table/telephony")
options = {"numPlans": 50, "numCustomers": 100}

dfPlans = multiTableDS.get(table="plans", **options).build()
dfCustomers = multiTableDS.get(table="customers", **options).build()
dfDeviceEvents = multiTableDS.get(table="deviceEvents", **options).build()

dfInvoices = multiTableDS.getSummaryDataset(table="invoices",
                                             plans=dfPlans,
                                             customers=dfCustomers,
                                             deviceEvents=dfDeviceEvents)
display(dfInvoices)
```

--------------------------------

### Simulating returns with a Gamma distribution for return delay

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/DISTRIBUTIONS.md

This example demonstrates how to use a Gamma distribution to favor shorter return delays when generating return dates.

```python
from pyspark.sql.types import IntegerType

import dbldatagen as dg
import dbldatagen.distributions as dist


row_count = 1000 * 100
testDataSpec = (
    dg.DataGenerator(spark, name="test_data_set1", rows=row_count)
    .withColumn("purchase_id", IntegerType(), minValue=1000000, maxValue=2000000)
    .withColumn("product_code", IntegerType(), uniqueValues=10000, random=True)
    .withColumn(
        "purchase_date",
        "date",
        data_range=dg.DateRange("2017-10-01 00:00:00", "2018-10-06 11:55:00", "days=3"),
        random=True,
    )
    # create return delay , favoring short delay times
    .withColumn(
        "return_delay",
        "int",
        minValue=1,
        maxValue=100,
        random=True,
        distribution=dist.Gamma(1.0, 2.0),
        omit=True,
    )
    .withColumn(
        "return_date",
        "date",
        expr="date_add(purchase_date, return_delay)",
        baseColumn=["purchase_date", "return_delay"],
    )
)

dfTestData = testDataSpec.build()
```

--------------------------------

### Example 1: Site code and technology

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_streaming_data.rst

This code snippet demonstrates how to generate streaming data for site information, including site ID, site code, and sector technology. It utilizes `withStreaming=True` and specifies `rowsPerSecond` to control the data generation rate.

```python
from datetime import timedelta, datetime
import math
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, \
                                 FloatType, TimestampType
# from dbldatagen.data_generator import DataGenerator,ensure
import dbldatagen as dg

interval = timedelta(days=1, hours=1)
start = datetime(2017, 10, 1, 0, 0, 0)
end = datetime(2018, 10, 1, 6, 0, 0)

schema = StructType([
    StructField("site_id", IntegerType(), True),
    StructField("site_cd", StringType(), True),
    StructField("c", StringType(), True),
    StructField("c1", StringType(), True),
    StructField("sector_technology_desc", StringType(), True),
])

# will have implied column `id` for ordinal of row
ds = (
    dg.DataGenerator(spark, name="association_oss_cell_info", rows=100000, partitions=20)
    .withSchema(schema)
    # withColumnSpec adds specification for existing column
    .withColumnSpec("site_id", minValue=1, maxValue=20, step=1)
    # base column specifies dependent column
    .withIdOutput()
    .withColumnSpec("site_cd", prefix='site', baseColumn='site_id')
    .withColumn("sector_status_desc", "string", minValue=1, maxValue=200, step=1,
                prefix='status', random=True)
    # withColumn adds specification for new column
    .withColumn("rand", "float", expr="floor(rand() * 350) * (86400 + 3600)")
    .withColumn("last_sync_dt", "timestamp", begin=start, end=end, interval=interval,
                random=True)
    .withColumnSpec("sector_technology_desc", values=["GSM", "UMTS", "LTE", "UNKNOWN"],
                random=True)
    .withColumn("test_cell_flg", "integer", values=[0, 1], random=True)
)

df = ds.build(withStreaming=True, options={'rowsPerSecond': 500})

display(df)

```

--------------------------------

### Loading DataGenerator from JSON

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/serialized_data_generators.rst

Illustrates how to create a DataGenerator instance from a JSON string using `DataGenerator.loadFromJson()`.

```python
import dbldatagen as dg

# Define the data generation options:
jsonStr = '''{
   "name": "users_dataset",
   "rows": 1000,
   "randomSeedMethod": "hash_fieldname",
   "columns": [
      {"colName": "user_name", "colType": "string", "expr": "concat('user_', id)"},
      {"colName": "phone_number", "colType": "string", "template": "555-DDD-DDDD"}
   ]
}'''

# Create a data generator from the JSON string:
testDataSpec = dg.DataGenerator.loadFromJson(jsonStr)
```

--------------------------------

### Generating Complex Column Data Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/generating_json_data.rst

This example demonstrates how to generate columns with complex data types such as arrays, structs, and maps, using dbldatagen.

```python
import dbldatagen as dg

ds = (
     dg.DataGenerator(spark, name="test_data_set1", rows=1000, random=True)
    .withColumn("r", "float", minValue=1.0, maxValue=10.0, step=0.1,
                numColumns=5)
    .withColumn("observations", "array<float>",
                expr="slice(array(r_0, r_1, r_2, r_3, r_4), 1, abs(hash(id)) % 5 + 1 )",
                baseColumn="r")
    )

df = ds.build()
df.show()
```

--------------------------------

### Device Data Generation Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/generating_json_data.rst

This example demonstrates how to generate a dataset of device information, including various data types and complex structures, using dbldatagen.

```python
import dbldatagen as dg

device_population = 100000

country_codes = ['CN', 'US', 'FR', 'CA', 'IN', 'JM', 'IE', 'PK', 'GB', 'IL', 'AU', 'SG',
                 'ES', 'GE', 'MX', 'ET', 'SA', 'LB', 'NL']
country_weights = [1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 126, 109, 58, 8,
                  17]

manufacturers = ['Delta corp', 'Xyzzy Inc.', 'Lakehouse Ltd', 'Acme Corp', 'Embanks Devices']

lines = ['delta', 'xyzzy', 'lakehouse', 'gadget', 'droid']

testDataSpec = (
    dg.DataGenerator(spark, name="device_data_set", rows=1000000,
                     partitions=8,
                     randomSeedMethod='hash_fieldname')
    .withIdOutput()
    # we'll use hash of the base field to generate the ids to
    # avoid a simple incrementing sequence
    .withColumn("internal_device_id", LongType(), minValue=0x1000000000000,
                uniqueValues=device_population, omit=True, baseColumnType="hash")

    # note for format strings, we must use "%lx" not "%x" as the
    # underlying value is a long
    .withColumn("device_id", StringType(), format="0x%013x",
                baseColumn="internal_device_id")

    # the device / user attributes will be the same for the same device id
    # so lets use the internal device id as the base column for these attribute
    .withColumn("country", StringType(), values=country_codes,
                weights=country_weights,
                baseColumn="internal_device_id")

    .withColumn("manufacturer", StringType(), values=manufacturers,
                baseColumn="internal_device_id", omit=True)
    .withColumn("line", StringType(), values=lines, baseColumn="manufacturer",
                baseColumnType="hash", omit=True)
    .withColumn("manufacturer_info", "string",
                expr="to_json(named_struct('line', line, 'manufacturer', manufacturer))",
                baseColumn=['manufacturer', 'line'])


    .withColumn("model_ser", IntegerType(), minValue=1, maxValue=11,
                baseColumn="device_id",
                baseColumnType="hash", omit=True)

    .withColumn("event_type", StringType(),
                values=["activation", "deactivation", "plan change",
                        "telecoms activity", "internet activity", "device error"],
                random=True, omit=True)
    .withColumn("event_ts", "timestamp",
                begin="2020-01-01 01:00:00", end="2020-12-31 23:59:00",
                interval="1 minute", random=True, omit=True)

    .withColumn("event_info", "string",
                expr="to_json(named_struct('event_type', event_type, 'event_ts', event_ts))",
                baseColumn=['event_type', 'event_ts'])
    )

dfTestData = testDataSpec.build()

#dfTestData.write.format("json").mode("overwrite").save("/tmp/jsonData2")
display(dfTestData)
```

--------------------------------

### Saving data generation specs to configuration

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/serialized_data_generators.rst

Demonstrates how to convert a DataGenerator object to a Python dictionary using the saveToInitializationDict method.

```python
from pyspark.sql.types import StringType
import dbldatagen as dg

# Create a sample data generator with a few columns:
testDataSpec = (
    dg.DataGenerator(spark, name="users_dataset", rows=1000)
    .withColumn("user_name", StringType(), expr="concat('user_', id)")
    .withColumn("email_address", StringType(), expr="concat(user_name, '@email.com')")
    .withColumn("phone_number", StringType(), template="555-DDD-DDDD")
)

# Get the data generation options as a Python dictionary:
dataSpecOptions = testDataSpec.saveToInitializationDict()
```

--------------------------------

### Device Data Generation Example

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/generating_json_data.rst

This code snippet shows a comprehensive example of generating device data with various column types, including IDs, country, manufacturer, model, event types, and timestamps, using dbldatagen.

```python
country_weights = [1300, 365, 67, 38, 1300, 3, 7, 212, 67, 9, 25, 6, 47, 83, 126, 109, 58, 8,
                      17]

manufacturers = ['Delta corp', 'Xyzzy Inc.', 'Lakehouse Ltd', 'Acme Corp', 'Embanks Devices']

lines = ['delta', 'xyzzy', 'lakehouse', 'gadget', 'droid']

testDataSpec = (
       dg.DataGenerator(spark, name="device_data_set", rows=1000000, 
                        partitions=8, randomSeedMethod='hash_fieldname')
       .withIdOutput()
       # we'll use hash of the base field to generate the ids to
       # avoid a simple incrementing sequence
       .withColumn("internal_device_id", LongType(), minValue=0x1000000000000,
                   uniqueValues=device_population, omit=True, baseColumnType="hash")

       # note for format strings, we must use "%lx" not "%x" as the
       # underlying value is a long
       .withColumn("device_id", StringType(), format="0x%013x", 
                   baseColumn="internal_device_id")

       # the device / user attributes will be the same for the same device id
       # so lets use the internal device id as the base column for these attribute
       .withColumn("country", StringType(), values=country_codes,
                   weights=country_weights,
                   baseColumn="internal_device_id")

       .withColumn("manufacturer", StringType(), values=manufacturers,
                   baseColumn="internal_device_id", omit=True)
       .withColumn("line", StringType(), values=lines, baseColumn="manufacturer",
                   baseColumnType="hash", omit=True)
       .withColumn("manufacturer_info", StructType([StructField('line',StringType()),
                                                   StructField('manufacturer', StringType())]),
                   expr="named_struct('line', line, 'manufacturer', manufacturer)",
                   baseColumn=['manufacturer', 'line'])


       .withColumn("model_ser", IntegerType(), minValue=1, maxValue=11,
                   baseColumn="device_id",
                   baseColumnType="hash", omit=True)

       .withColumn("event_type", StringType(),
                   values=["activation", "deactivation", "plan change",
                           "telecoms activity", "internet activity", "device error"],
                   random=True, omit=True)
       .withColumn("event_ts", "timestamp", begin="2020-01-01 01:00:00",
                   end="2020-12-31 23:59:00",
                   interval="1 minute", random=True, omit=True)

       .withColumn("event_info",
                    StructType([StructField('event_type',StringType()),
                                StructField('event_ts', TimestampType())]),
                   expr="named_struct('event_type', event_type, 'event_ts', event_ts)",
                   baseColumn=['event_type', 'event_ts'])
       )

   dfTestData = testDataSpec.build()
   dfTestData.write.format("json").mode("overwrite").save("/tmp/jsonData2")
```

--------------------------------

### Listing Available Datasets

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_standard_datasets.rst

Lists available datasets that match a pattern and support streaming.

```python
import dbldatagen as dg
dg.Datasets.list(pattern="basic.*", supportsStreaming=True)
```

--------------------------------

### Getting Details of a Dataset

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_standard_datasets.rst

Describes a specific standard dataset, in this case, 'basic/user'.

```python
import dbldatagen as dg
dg.Datasets.describe("basic/user")
```

--------------------------------

### Saving DataGenerator to JSON

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/serialized_data_generators.rst

Demonstrates how to create a DataGenerator and save its configuration to a JSON string using `saveToJson()`.

```python
from pyspark.sql.types import StringType
import dbldatagen as dg

# Create a sample data generator with a few columns:
testDataSpec = (
    dg.DataGenerator(spark, name="users_dataset", rows=1000)
    .withColumn("user_name", StringType(), expr="concat('user_', id)")
    .withColumn("email_address", StringType(), expr="concat(user_name, '@email.com')")
    .withColumn("phone_number", StringType(), template="555-DDD-DDDD")
)

# Create a JSON string with the data generation config:
jsonStr = testDataSpec.saveToJson()
```

--------------------------------

### Streaming Data to Delta Table

Source: https://github.com/databrickslabs/dbldatagen/blob/master/docs/source/using_streaming_data.rst

Example of generating streaming data and writing it to a Delta table.

```python
from datetime import timedelta, datetime
import dbldatagen as dg

interval = timedelta(days=1, hours=1)
start = datetime(2017, 10, 1, 0, 0, 0)
end = datetime(2018, 10, 1, 6, 0, 0)

# row count will be ignored
ds = (dg.DataGenerator(spark, name="association_oss_cell_info", rows=100000, partitions=20)
      .withColumnSpec("site_id", minValue=1, maxValue=20, step=1)
      .withColumnSpec("site_cd", prefix='site', baseColumn='site_id')
      .withColumn("sector_status_desc", "string", minValue=1, maxValue=200, step=1,
                  prefix='status', random=True)
      .withColumn("rand", "float", expr="floor(rand() * 350) * (86400 + 3600)")
      .withColumn("last_sync_dt", "timestamp", begin=start, end=end, interval=interval,
                  random=True)
      .withColumn("sector_technology_desc", values=["GSM", "UMTS", "LTE", "UNKNOWN"],
                  random=True)
      )

df = ds.build(withStreaming=True, options={'rowsPerSecond': 500})

df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/tmp/dbldatagen/streamingDemo/checkpoint")
    .start("/tmp/dbldatagen/streamingDemo/data")
```