KeesCBakker

bucket = "/mnt/{}{}-delta".format(bucket_prefix, topic)

delta_location = bucket + "/delta-table"
checkpoint_location = bucket + "/checkpoints";
schema_location = bucket + "/kafka_schema.json";

if debug:
  delta_location += "_debug"
  checkpoint_location += "_debug"
  schema_location += "_debug.json" 

KeesCBakker

debug = str_to_bool(getArgument("debug"))
topic = validate_required_argument_and_return_value("topic")
streaming = str_to_bool(getArgument("streaming"))
update_kafka_schema = str_to_bool(getArgument("update_kafka_schema"))

KeesCBakker

def str_to_bool(value):
  FALSE_VALUES = ['false', 'no', '0']
  TRUE_VALUES = ['true', 'yes', '1']
  lvalue = str(value).lower()
  if lvalue in (FALSE_VALUES): return False
  if lvalue in (TRUE_VALUES):  return True
  raise Exception("String value should be one of {}, but got '{}'.".format(
    FALSE_VALUES + TRUE_VALUES, value))

KeesCBakker

true_false = ["True", "False"]

dbutils.widgets.removeAll()
dbutils.widgets.text("topic", "", "Topic")
dbutils.widgets.dropdown("debug", "True", true_false, "Debug")
dbutils.widgets.dropdown("streaming", "False", true_false, "Streaming")
dbutils.widgets.dropdown("update_kafka_schema", "False", true_false, "Update Kafka Schema")

KeesCBakker

kafka_broker = "my.kafka.queue:9092"
bucket_prefix = "my-company-bucket-prefix-"

KeesCBakker

public static class SortingExtensions
{
    public static IDictionary<TKey, TValue> SortBy<TKey, TValue>(
        this IDictionary<TKey, TValue> dictionary, 
        IEnumerable<TKey> keys
    )
    {
        var sorter = new KeyComparer<TKey>(keys);
        return new SortedDictionary<TKey, TValue>(dictionary, sorter);
    }

KeesCBakker

public class KeyComparer<T> : IComparer<T>
{
    private const int TOP = -1;
    private const int BOTTOM = 1;
    private const int EQUAL = 0;

    private readonly Dictionary<T, int> _keys = new Dictionary<T, int>();

    public KeyComparer(IEnumerable<T> keys)
    {

KeesCBakker

public class KeyComparer<T> : IComparer<T>
{
    private const int TOP = -1;
    private const int BOTTOM = 1;
    private const int EQUAL = 0;

    private readonly Dictionary<T, int> _keys = new Dictionary<T, int>();

    public KeyComparer(IEnumerable<T> keys)
    {

KeesCBakker

const UpdatableMessage = class {

  constructor(token, channel) {
    this.token = token;
    this.channel = channel;
    this.ts = null;
    this.message = null;
    this.nextMessage = null;
    this.sending = false;
  }

KeesCBakker

const axios = require("axios");

function sendMessage(token, channel, ts, msg) {

  token = encodeURIComponent(token);
  channel = encodeURIComponent(channel);
  msg = encodeURIComponent(msg);

  const action = ts ? "update" : "postMessage";
  const url = `https://slack.com/api/chat.${action}?token=${token}&channel=${channel}&text=${msg}&as_user=true&ts=${ts}`;