PREFECT__LOGGING__EXTRA_LOGGERS

import hydra
from omegaconf import DictConfig, OmegaConf
import os

import uuid

# @hydra.main(config_path="conf", config_name="config")
from hydra import compose, initialize
from omegaconf import OmegaConf


# print(os.environ)
# print(subprocess.run(["ls"]))
import os
import warnings
import sys
from funcy.colls import project
import mlflow
import subprocess
import prefect
import logging
import sys


def set_env(cfg: DictConfig) -> None:
    """
    set all env var
    """
    env = cfg["var"]
    for k, v in env.items():
        os.environ[k] = v


initialize(config_path="conf", job_name="gojob")
cfg = compose(config_name="config")
print(OmegaConf.to_yaml(cfg))
traking = cfg["var"]["MLFLOW_TRACKING_URI"]
params = {}
project_path = cfg["project_path"]
experiment = cfg["experiment"]
mlflow.set_tracking_uri(traking)
mlflow.set_experiment(experiment)
l = "mlflow"
logger = logging.getLogger(l)
logger.setLevel("INFO")
log_stream = logging.StreamHandler(sys.stdout)
LOG_FORMAT = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
log_stream.setFormatter(LOG_FORMAT)
logger.addHandler(log_stream)
# print(subprocess.run(["ls"]))
with mlflow.start_run(nested=True):

    set_env(cfg)

    <http://logger.info|logger.info>("---------test------------")

    <http://logger.info|logger.info>("-----------test2-----------")

    # logger = prefect.context.get("logger")
    # <http://logger.info|logger.info>("------------An MLFLOW RUN ------------------")
    get_data = mlflow.run(project_path, "process_data", experiment_name=experiment)

    train = mlflow.run(project_path, "train", experiment_name=experiment)

import mlflow
from mlflow import projects
import os
from prefect import Flow
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
from prefect.run_configs import LocalRun
import prefect
from prefect import task, Flow, Parameter
import subprocess
from prefect import Client
import hydra
from omegaconf import DictConfig, OmegaConf
from dataclasses import dataclass
from omegaconf import DictConfig, OmegaConf
from hydra.core.config_store import ConfigStore
import yaml
import logging
import sys

# os.environ["AWS_SECRET_ACCESS_KEY"] = "admin1598753"
# os.environ["AWS_ACCESS_KEY_ID"] = "adminminio"
# os.environ["MLFLOW_S3_ENDPOINT_URL"] = "<http://localhost:3001>"
# os.environ["MLFLOW_S3_IGNORE_TLS"] = "true"
# os.environ["MLFLOW_TRACKING_URI"] = "localhost:3000"
# print("os")
# project_path = "./project"
# experiment = "gojob"
# traking = "<http://localhost:5000>"
# params = {}


@task
def set_env(cfg: DictConfig) -> None:
    """
    set all env var
    """
    env = cfg["var"]
    for k, v in env.items():
        os.environ[k] = v


@task
def set_uri(traking):
    mlflow.set_tracking_uri(traking)
    return traking


@task(log_stdout=True)
def set_exp(experiment):
    print("log set experiment ")
    mlflow.set_experiment(experiment)
    return experiment


@task()
def run_mlflow(project_path, experiment):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("------------An info message.------------------")
    l = "mlflow"
    logger = logging.getLogger(l)
    logger.setLevel("INFO")
    log_stream = logging.StreamHandler(sys.stdout)
    LOG_FORMAT = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    log_stream.setFormatter(LOG_FORMAT)
    logger.addHandler(log_stream)
    mlflow.projects.run(
        project_path, experiment_name=experiment,
    )


print("experiment setted")


@dataclass
class Run:
    flow_id: str = "None"


@hydra.main(config_path="project/conf", config_name="config")
def workflow(cfg: DictConfig):
    with Flow("gojobflow", run_config=LocalRun()) as flow:
        logger = prefect.context.get("logger")
        logger = prefect.utilities.logging.get_logger()
        experiment = cfg["experiment"]
        tracking = cfg["var"]["MLFLOW_TRACKING_URI"]
        project_path = cfg["project_path"]

        v = set_env(cfg)
        s = set_uri(tracking)
        e = set_exp(experiment)
        r = run_mlflow(project_path, e)
    try:
        idf = flow.register(project_name="gojob", set_schedule_active=False)
        run_1 = {"flow_id": idf}
        <http://logger.info|logger.info>(cfg["project_path"] + "/conf/run/run_1.yaml")
        with open(cfg["project_path"] + "/conf/run/run_1.yaml", "w+") as outfile:
            yaml.dump(run_1, outfile, default_flow_style=False)
    except:
        subprocess.run(["prefect", "create", "project", "gojob"])
        idf = flow.register(project_name="gojob", set_schedule_active=False)
        run_1 = {"flow_id": idf}
        <http://logger.info|logger.info>(cfg["project_path"] + "/conf/run/run_1.yaml")
        with open(cfg["project_path"] + "/conf/run/run_1.yaml", "w+") as outfile:
            yaml.dump(run_1, outfile, default_flow_style=False)

    # ri = Run(flow_id=idf)
    # Registering the Config class with the name 'config'.
    # cs.store(group="run", name="run_1", node=ri)
    # print(OmegaConf.to_yaml(cfg))


if __name__ == "__main__":
    workflow()

export PREFECT__LOGGING__EXTRA_LOGGERS="['boto3', 'botocore']"
export PREFECT__LOGGING__LEVEL="DEBUG"

from prefect import task, Flow
import boto3
import pandas as pd
import prefect
import logging

@task
def get_s3_data():
    logging.getLogger("boto3").setLevel(logging.DEBUG)
    logging.getLogger("botocore").setLevel(logging.DEBUG)
    prefect_logger = prefect.context.get("logger")
    s3 = boto3.client("s3")
    obj = s3.get_object(
        Bucket="data-lake-bronze",
        Key="global_power_plant_dataset/global_power_plant_database.csv",
    )
    df = pd.read_csv(obj["Body"])
    <http://prefect_logger.info|prefect_logger.info>("Read df: %s", df.head(2))

with Flow("log_test",) as flow:
    get_s3_data()

import redis
import cloudpickle
import pickle
import prefect
from prefect import task, Flow, Parameter
import subprocess

# multi key usage


import prefect
from prefect import task, Flow
import cloudpickle


@task
def set_print():
    r = redis.Redis(host="localhost", port=6379, db=0)
    r.set("foo", "bar")
    print("regular key ", r.get("foo"))


@task
def redis_dict():
    user = {
        "Name": "Pradeep",
        "Company": "SCTL",
        "Address": "Mumbai",
        "Location": "RCP",
    }
    r = redis.Redis(host="localhost", port=6379, db=0)
    r.hmset("pythonDict", user)

    print(r.hgetall("pythonDict"))


@task
def hello_task():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello world!")
    r = redis.Redis(host="localhost", port=6379, db=0)
    r.set("logger", cloudpickle.dumps(logger))
    subprocess.run(["python", "task_subprocess.py"])


with Flow("hello-flow") as flow:
    hello_task()

flow.run()

import pickle
import redis
print('start')
r = redis.Redis(host="localhost", port=6379, db=0)

logger = pickle.loads(r.get("logger"))
<http://logger.info|logger.info>("It is working!")
print("runned")

prefect

@task(log_stdout=True)
def run_mlflow(project_path, experiment):
    print("mlflowrun")
    logger = prefect.context
    r = redis.Redis(host="localhost", port=6379, db=0)
    print(" pinckling ")
    r.set("context", cloudpickle.dumps(logger))
    print( "" )
    subprocess.run(["python", "redis/task_subprocess.py"])
    mlflow.projects.run(
        project_path,
        experiment_name=experiment,
    )

import pickle
import redis
from prefect.utilities import logging

print("start")
handler = logging.CloudHandler()
r = redis.Redis(host="localhost", port=6379, db=0)
logger = pickle.loads(r.get("context")).get("logger")
logger.addHandler(handler)
<http://logger.info|logger.info>("+++++++++++++++++++++++++")
<http://logger.info|logger.info>("It is working!")
print("regular key ", r.get("foo"))
print(logger)
print("runned")

Context