Convert45

CHANGELOG.md

@@ -1,21 +1,16 @@
-0.4.6 (unreleased)
+0.5.0 (unreleased)
 ------------------
 
-- Nothing changed yet.
+- Use extrinsic folder names for project/package
+- do: schedule for top-level ordering and scheduling
 
 
 0.4.5 (2022-05-10)
 ------------------
 
 - Fix Manifest
-
-
-0.4.4 (unreleased)
-------------------
-
 - YMM 0.6.1 support: open direct URL for test runs
 
-
 0.4.3 (2022-04-18)
 ------------------
 

archive-v45/cqml45.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+import os, yaml
+
+#
+# Take two paths: root4 root5
+# Copy files in each subfolder
+# Strip out meta
+# Strip leading numbers
+#
+
+ROOT="/Users/nauto/Developer"
+R4=f"{ROOT}/it/databricks"
+R5=f"{ROOT}/dbt/cqml"
+
+def read_yaml(yaml_file):
+    with open(yaml_file) as data:
+        raw_yaml = yaml.full_load(data)
+        return raw_yaml
+
+def write_yaml(yaml_file, raw_yaml):
+    with open(yaml_file, 'w') as file:
+        yaml.dump(raw_yaml, file, sort_keys=False)
+
+def extract(root):
+    tree = []
+    for folder in os.scandir(root):
+        if folder.is_dir():
+            print(folder.name)
+            for file in os.scandir(folder.path):
+                if file.name.endswith(".yml"):
+                    yml = read_yaml(file.path)
+                    node = {"file":file.name,"project":folder.name, "yml": yml}
+                    tree.append(node)
+    return tree
+
+def convert(root, node):
+    dir = node["project"]
+    file = ''.join([c for c in node["file"] if not c.isdigit()])
+    prefix = file.split("_")[0]
+    name = file.split("_")[1]
+    if prefix == "rnr": dir = prefix
+    if prefix == "sierra": name = f"{prefix}.yml"
+    path = os.path.join(root, dir, name)
+    yml = node["yml"]
+    del yml["meta"]
+    yml["cqml"] = 0.5
+    yml["project"] = dir
+    yml["package"] = name
+    write_yaml(path, yml)
+    return path
+
+t = extract(R4)
+print(f"\nExtracted: {len(t)} files\n")
+#print(t[0]["yml"])
+for n in t:
+    print(n["file"])
+    p = convert(R5, n)
+    print("\t",p)

pipes/all.yml

@@ -0,0 +1,16 @@
+cqml: 0.6
+order: 0.0
+id: main
+env:
+  org: nauto
+  bucket: biz-databricks-prod-reports
+  catalog: quilt
+  root: /dbfs/tmp
+  path: .
+actions:
+  all:
+    do: run
+    start: 0500
+    pipes:
+      - demo/demo
+      - test/cqml

tests/demo.yml → pipes/demo/demo.yml

@@ -1,11 +1,5 @@
-cqml: 0.4.4
+cqml: 0.5.0
 id: cqml_demo
-meta:
-    org: nauto
-    project: sangam
-    s3.bucket: biz-databricks-root-prod-us
-    catalog: quilt
-    root: /dbfs/tmp
 actions:
   /Workspace/Repos/[email protected]/cqml/data:
     do: loadfiles
@@ -43,7 +37,7 @@ actions:
         file_ext: csv
         expiration_date: '2200-02-02'
     cols:
-      letter: sort
+      better: sort
       dat: sort
   widget-report:
     do: report
@@ -62,11 +56,11 @@ actions:
       letter: tbd
   plus_operator:
     do: call
-    from: rename_merged
+    from: grouped
     operator: +
     args:
-        - next
-        - num
+        - text
+        - sum_num
   calc_quarters:
     do: calc
     from: plus_operator
@@ -77,8 +71,8 @@ actions:
         - $col
         - 4
     cols:
-      num: qnum
-      next: qnumb
+      sum_num: qnum
+      text: qnumb
       plus_operator: qdumb
   flagged:
     do: flag
@@ -130,13 +124,3 @@ actions:
     cols:
       letter: tbd
       text: tbd
-  aggregates:
-    do: group
-    from: $id
-    agg:
-      concat_space: count
-      call_coalesce: sum
-    sort: n_concat_space
-    cols:
-      num: tbd
-      letter: tbd

tests/cqml_test.yml → pipes/test/cqml.yml

@@ -1,11 +1,5 @@
 cqml: 0.2
 id: cqml_test
-meta:
-    org: nauto
-    project: sangam
-    s3.bucket: biz-databricks-root-prod-us
-    catalog: quilt
-    root: /dbfs/tmp
 actions:
   /Workspace/Repos/[email protected]/cqml/data:
     do: loadfiles

src/cqml/__init__.py

@@ -4,4 +4,6 @@
 
 # TODO: https://github.com/LucaCanali/sparkMeasure
 
+from .yml import *
 from .wrappers import *
+from .root import Root

src/cqml/boxquilt.py

@@ -12,7 +12,7 @@
 from boxsdk import Client, OAuth2, JWTAuth
 from pyspark.sql import Row
 from pyspark.sql.functions import udf,lit
-from pyspark.sql.types import StringType
+from pyspark.sql.types import StringType, StructType
 import os
 
 def dir_row(folder):
@@ -119,7 +119,7 @@ def create_or_update_box(self, skipUpdate=False):
         dbfs = list(self.rows.keys())
         to_create = list(set(dbfs) - set(box))
         to_update = list(set(dbfs).intersection(box))
-        print(f"create:{len(to_create)} update:{len(to_update)}")
+        print(f"box_create:{len(to_create)} update:{len(to_update)}")
 
         n = 0
         for name in to_create:
@@ -147,5 +147,7 @@ def create_or_update_box(self, skipUpdate=False):
 
     def box_table(self):
         array = list(self.rows.values())
-        print(f'box_table: {len(array)}')
-        return self.spark.createDataFrame([Row(**i) for i in array])
+        if len(array) > 0:
+            print(f'box_table: {len(array)}')
+            return self.spark.createDataFrame([Row(**i) for i in array])
+        return self.spark.createDataFrame([], StructType([])) 

src/cqml/db2quilt.py

@@ -17,7 +17,6 @@
 from pathlib import Path
 import pprint
 pp = pprint.PrettyPrinter(indent=4)
-QPKG = q3.Package()
 
 def cleanup_names(df):
     for c in df.columns:
@@ -112,8 +111,8 @@ def callback(x): grid.restore(filter=[['{KEY}', '==', x]])
 grid
 """
 def make_widget(opts):
-    print('make_widget')
-    print(opts)
+    #print('make_widget')
+    #print(opts)
     code = [NB_WIDGET.format(KEY=col,WIDGET=w) for col, w in opts.items()]
     cells = [[True, c] for c in code]
     return cells
@@ -133,9 +132,17 @@ def make_slug(name): return re.sub(r'[^\w-]', '_', name.lower())
 Quilt Wrappers
 """
 
+DEFAULT_ENV={
+    'catalog': PKG_DIR,
+    'root': PYROOT,
+}
+
+def get_env(env, key, default):
+    return env[key] if key in env else default
+
 class Project:
     def __init__(self, config):
-        org, bucket, project = itemgetter('org','s3.bucket','project')(config)
+        org, bucket, project = itemgetter('org','bucket','project')(config)
         pkg_dir = config['catalog'] if 'catalog' in config else PKG_DIR
         root = config['root'] if 'root' in config else PYROOT
         self.repo = "s3://"+bucket
@@ -147,20 +154,21 @@ def package(self, id):
         return Package(id, self)
 
 class Package:
-    def __init__(self, id, proj, reset=False):
+    def __init__(self, id, proj, reset=True):
         self.id = id
         self.name = f"{proj.name}/{id}"
         self.proj = proj
         self.url = f"{proj.url}/{self.name}/"
         self.path = f"{proj.path}/{self.name}/"
         self.dir = to_dir(self.path)
+        self.pkg = q3.Package.browse(self.name, registry=self.proj.repo)
         if reset:
             shutil.rmtree(self.path,ignore_errors=True)
         make_dir(self.path)
         self.summaries={}
 
     def setup(self):
-        QPKG.install(self.name, registry=self.proj.repo, dest=self.path)
+        self.pkg.install(self.name, registry=self.proj.repo, dest=self.path)
 
     def read_csv(self, filename):
         path = self.path+filename
@@ -170,8 +178,10 @@ def read_csv(self, filename):
 
     def cleanup(self, msg, meta = {"db2quilt":"v0.1"}):
         self.write_summary()
-        QPKG.set_dir('/',path=self.path, meta=meta)
-        QPKG.push(self.name, self.proj.repo, message=msg,force=True) #,
+        self.pkg.set_dir('/',path=self.path, meta=meta)
+        print(self.pkg)
+
+        self.pkg.push(self.name, self.proj.repo, message=msg,force=True) #,
         #shutil.rmtree(self.path)
         self.html = f'Published <a href="{self.url}">{self.name}</a> for <b>{msg}</b>'
         return self
@@ -293,10 +303,13 @@ def write_summary(self):
 #
 
 def extract_pkg(cvm):
-    id, config = itemgetter('id','meta')(cvm.yaml)
+#    print(cvm.yaml)
+    config = itemgetter(kEnv)(cvm.yaml)
+    #print(f"extract_pkg.config: {config}")
     proj = Project(config)
+    id = config["package"]
     pkg_id = id + DEBUG_SUFFIX if cvm.debug == True else id
-    print("extract_pkg: "+pkg_id)
+    #print("extract_pkg: "+pkg_id)
     pkg = proj.package(pkg_id)
     #pkg.setup()
     return pkg

src/cqml/keys.py

@@ -9,6 +9,7 @@
 TRACE=True
 
 cAlias='|'
+kALL='_all_'
 kAbove='above'
 kAny='any'
 kArgs='args'
@@ -17,6 +18,7 @@
 kCount='count'
 kDoc='+doc'
 kDrop='drop'
+kEnv='env'
 kExt='file_ext'
 kFunc='function'
 kGroup='group'

src/cqml/root.py

@@ -0,0 +1,72 @@
+import os, yaml
+from .keys import *
+from .wrappers import CQML, pkg_cvm
+
+def read_yaml(yaml_file):
+    with open(yaml_file) as data:
+        raw_yaml = yaml.full_load(data)
+        return raw_yaml
+
+class Root:
+    def __init__(self, root):
+        self.root = root.split("/")[-1]
+        self.pipes = {}
+        self.env = {}
+        self.scan(root)
+
+    def keys(self): return list(self.pipes.keys())
+
+    def add_env(self, yml, key):
+        if not kEnv in yml: return {}
+        self.env[key] = yml[kEnv]
+        return yml[kEnv]
+
+    def set_env(self, yml, key):
+        print(f'set_env:{key}')
+        folder = yml[kEnv]["project"]
+        env = {}
+        if self.root in self.env: env.update(self.env[self.root])
+        if folder in self.env: env.update(self.env[folder])
+        if kEnv in yml: env.update(yml[kEnv])
+        yml[kEnv] = env
+        return env
+
+    def new(self, spark, key, debug=False):
+        pipe = self.pipes[key]
+        self.set_env(pipe, key)
+        cvm = CQML(pipe, spark)
+        if debug: cvm.debug = True
+        return cvm
+
+    def pkg(self, spark, key, debug=False):
+        cvm = self.new(spark, key, debug)
+        cvm.run();
+        return pkg_cvm(cvm)
+
+    #def pkg_all(self, spark, debug=False): return {key:self.pkg(spark, key, debug) for key in self.keys()}
+
+    def scan(self, root):
+        for entry in os.scandir(root):
+            self.parse(entry, root)
+
+    def parse(self, entry, folder):
+        name = entry.name
+        if name.endswith(".yml"):
+            yml = read_yaml(entry.path)
+            file_key = os.path.splitext(name)[0]
+            folder_key = folder.split("/")[-1]
+            env = self.add_env(yml, folder_key)
+            key = f"{folder_key}/{file_key}"
+            source = {
+                "file": name,
+                "package": file_key,
+                "project": folder_key,
+                "key": key,
+                "path": entry.path,
+            }
+            env.update(source)
+            yml[kEnv] = env
+            self.pipes[key] = yml
+        elif entry.is_dir():
+            print(entry.name)
+            self.scan(entry.path)

src/cqml/vm.py

@@ -73,6 +73,7 @@ def log(self, str, name=False):
       if self.debug:
          if name: print(name)
          print(str)
+      return str
 
     def macro(self, todo, action):
         mdef = todo.split("|")