skrawcz

"""
This module demonstrates a telephone application
using Burr that:
 - captions an image
 - creates caption embeddings (for analysis)
 - creates a new image based on the created caption
"""
import os
import uuid

skrawcz

# functions.py - declare and link your transformations as functions....
import pandas as pd
from hamilton.htypes import Parallelizable, Collect


def motor(motor_list: list[int]) -> Parallelizable[int]:
   for _motor in motor_list:
      yield _motor

def _is_motor_on(motor: int ) -> bool:

skrawcz

from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

# set the Airflow DAG parameters. This will appear in the Airflow UI. 
DEFAULT_DAG_PARAMS = dict(
    label="absenteeism_time_in_hours",
    feature_set=[
          "age_zero_mean_unit_variance",
          "has_children",
          "has_pet",

skrawcz

from hamilton.function_modifiers import extract_columns
import pandas as pd

async def _run_query(col_name: str, query_string: str) -> pd.DataFrame:
    # this would go to the database -- ideally the client is passed in as a parameter
    # the assumption here is that the database driver is asyncio based, else there's no
    # value in doing this :)
    return pd.DataFrame({col_name: [query_string]})

# async Hamilton func for query #1

skrawcz

from hamilton import driver
import transforms

# load the client
kafka_client = KafkaClient() # or whatever


config = {...}
dr = driver.Driver(config, transforms, adapter=...)

skrawcz

import pandas as pd

def total_distance_travelled_by_train( ... ) -> pd.Series:
    return # placeholder -- fill with actual logic, etc.
    
def distance_travelled_till_intermediate_station( ... ) -> pd.Series:
    return # placeholder -- fill with actual logic, etc.


def journey_distance(total_distance_travelled_by_train: pd.Series, distance_travelled_till_intermediate_station: pd.Series) -> pd.Series:

skrawcz

from typing import Tuple
from hamilton import base

class PandasDFWithDebugResultBuilder(base.ResultMixin):
    """This class is an example to show how you can extend the result building functionality of Hamilton.
    
    This result builder returns a dataframe, and dictionary of outputs that we don't want to make into a dataframe, 
    but are useful for debugging (for example). Caveat: this wont work for ray, dask, or spark usage without some tweaks.
    
    Example Usage::

skrawcz

from typing import Any

import pandas as pd

"""
Notes:
  1. This file is used for all the [ray|dask|spark]/hello_world examples.
  2. It therefore show cases how you can write something once and not only scale it, but port it
     to different frameworks with ease!
"""

skrawcz

@tag(owner='Data-Science', pii='False')
@check_output(data_type=np.float64, range=(-5.0, 5.0), allow_nans=False)
def height_zero_mean_unit_variance(height_zero_mean: pd.Series,
                                   height_std_dev: pd.Series) -> pd.Series:
   """Zero mean unit variance value of height"""
   return height_zero_mean / height_std_dev

skrawcz

# --- my_functions.py
import pandas as pd


def avg_3wk_spend(spend: pd.Series) -> pd.Series:
    """Rolling 3 week average spend."""
    return spend.rolling(3).mean()


def spend_per_signup(spend: pd.Series, signups: pd.Series) -> pd.Series: