# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# + jupyter={"source_hidden": true}
schema_raw = {'type': 'struct',
 'fields': [{'name': 'index',
   'type': {'type': 'struct',
    'fields': [{'name': '_index',
      'type': 'string',
      'nullable': True,
      'metadata': {}},
     {'name': '_type', 'type': 'string', 'nullable': True, 'metadata': {}}]},
   'nullable': True,
   'metadata': {}},
  {'name': 'layers',
   'type': {'type': 'struct',
    'fields': [{'name': 'cflow',
      'type': {'type': 'struct',
       'fields': [{'name': 'cflow_cflow_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_od_id',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_sequence',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_sequence_analysis_expected_sn',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_sequence_analysis_previous_frame',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_timestamp',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_cflow_version',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_sequence__ws_expert',
         'type': {'type': 'struct',
          'fields': [{'name': '_ws_expert__ws_expert_group',
            'type': 'string',
            'nullable': True,
            'metadata': {}},
           {'name': '_ws_expert__ws_expert_message',
            'type': 'string',
            'nullable': True,
            'metadata': {}},
           {'name': '_ws_expert__ws_expert_severity',
            'type': 'string',
            'nullable': True,
            'metadata': {}},
           {'name': '_ws_expert_cflow_unexpected_sequence_number',
            'type': 'string',
            'nullable': True,
            'metadata': {}}]},
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_ack',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_fin',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_psh',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_reserved',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_rst',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_syn',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_tcpflags_cflow_tcpflags_urg',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_text',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_timedelta_cflow_timeend',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_timedelta_cflow_timestart',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'cflow_timestamp_cflow_exporttime',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_direction',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_dstaddr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_dstmac',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_dstport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_flows',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_flowset_id',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_flowset_length',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_inputint',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_ip_version',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_mpls_label_length',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_octets',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_outputint',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_packets',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_post_dstmac',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_post_srcmac',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_post_vlanid',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_protocol',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_sampler_id',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_srcaddr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_srcmac',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_srcport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_tcpflags',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_template_frame',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_timedelta',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_tos',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_cflow_vlanid',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_text',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'dns',
      'type': {'type': 'struct',
       'fields': [{'name': 'dns_dns_count_add_rr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_count_answers',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_count_auth_rr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_count_queries',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_flags',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_id',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_dns_unsolicited',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_authenticated',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_authoritative',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_checkdisable',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_opcode',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_rcode',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_recavail',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_recdesired',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_response',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_truncated',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_flags_dns_flags_z',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'dns_text',
         'type': {'type': 'array',
          'elementType': 'string',
          'containsNull': True},
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_a',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_cname',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_count_labels',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_qry_class',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_qry_name',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_qry_name_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_qry_type',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_resp_class',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_resp_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_resp_name',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_resp_ttl',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_dns_resp_type',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'text_text',
         'type': {'type': 'array',
          'elementType': 'string',
          'containsNull': True},
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'eth',
      'type': {'type': 'struct',
       'fields': [{'name': 'eth_dst_eth_addr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_dst_eth_addr_resolved',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_dst_eth_dst_resolved',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_dst_eth_ig',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_dst_eth_lg',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_eth_dst',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_eth_src',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_eth_type',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_src_eth_addr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_src_eth_addr_resolved',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_src_eth_ig',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_src_eth_lg',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'eth_src_eth_src_resolved',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'frame',
      'type': {'type': 'struct',
       'fields': [{'name': 'frame_frame_cap_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_encap_type',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_ignored',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_marked',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_number',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_offset_shift',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_protocols',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_time',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_time_delta',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_time_delta_displayed',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_time_epoch',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'frame_frame_time_relative',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'ip',
      'type': {'type': 'struct',
       'fields': [{'name': 'ip_dsfield_ip_dsfield_dscp',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_dsfield_ip_dsfield_ecn',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_flags_ip_flags_df',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_flags_ip_flags_mf',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_flags_ip_flags_rb',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_flags_ip_frag_offset',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_addr',
         'type': {'type': 'array',
          'elementType': 'string',
          'containsNull': True},
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_checksum',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_checksum_status',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_dsfield',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_dst',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_dst_host',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_flags',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_hdr_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_host',
         'type': {'type': 'array',
          'elementType': 'string',
          'containsNull': True},
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_id',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_len',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_proto',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_src',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_src_host',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_ttl',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'ip_ip_version',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'pflog',
      'type': {'type': 'struct',
       'fields': [{'name': 'pflog_pflog_action',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_af',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_daddr_ipv4',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_dir',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_dport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_ifname',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_length',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_pad',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_pid',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_reason',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_rewritten',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_rule_pid',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_rule_uid',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_rulenr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_ruleset',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_saddr_ipv4',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_sport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_subrulenr',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'pflog_pflog_uid',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}},
     {'name': 'udp',
      'type': {'type': 'struct',
       'fields': [{'name': 'udp_udp_checksum',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_checksum_status',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_dstport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_length',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_port',
         'type': {'type': 'array',
          'elementType': 'string',
          'containsNull': True},
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_srcport',
         'type': 'string',
         'nullable': True,
         'metadata': {}},
        {'name': 'udp_udp_stream',
         'type': 'string',
         'nullable': True,
         'metadata': {}}]},
      'nullable': True,
      'metadata': {}}]},
   'nullable': True,
   'metadata': {}},
  {'name': 'timestamp', 'type': 'string', 'nullable': True, 'metadata': {}}]}

# +
import nest_asyncio
import asyncio
import concurrent.futures

import socket
import json
from json import JSONDecodeError
import datetime

import tempfile
import shutil

import IPython.display

import pyspark
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import Row, SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as func

nest_asyncio.apply()

def stream_packets(_parse):
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.bind(("0.0.0.0", 22055))
        while True:
            p, _ = s.recvfrom(65536)
            parsed = ""
            while parsed == "":
                try:
                    parsed = _parse(p)
                except JSONDecodeError:
                    parsed = ""
                    pn, _ = s.recvfrom(65536)
                    p += pn
            yield parsed

def parse_message(x):
    s = x.decode("utf-8").rstrip()
    return [json.loads(m) for m in s.split("\n")]

def record_packets(data_dir):
    streamer = iter(stream_packets(parse_message))
    while True:
        start = datetime.datetime.now()
        now = str(int(start.timestamp()))
        with tempfile.NamedTemporaryFile(dir="/tmp", delete=False, mode="w") as f:
            fname = f.name
            i = 0
            while i < 10 and (datetime.datetime.now() - start).total_seconds() < 5:
                p = next(streamer)
                for m in p:
                    f.write(json.dumps(m) + "\n")
        shutil.move(fname, data_dir + "/" + now + ".jsonl")

spark = SparkSession.builder\
    .appName("Read Socket")\
    .master("local")\
    .enableHiveSupport()\
    .getOrCreate()

sc = spark.sparkContext
ssc = StreamingContext(sc, 1)

schema = StructType.fromJson(schema_raw)


# +
class DnsRecordOperator:
    
    def __getstate__(self):
        self._ncalled["__getstate__"] += 1
        state = self.__dict__.copy()
        state["_hostnames_updated_str"] = state["_hostnames_updated"].isoformat()
        del state["_hostnames_updated"]
        return state
    
    def __setstate__(self, state):
        state["_hostnames_updated"] = datetime.datetime.fromisoformat(
            state["_hostnames_updated_str"]
        )
        del state["_hostnames_updated_str"]
        self.__dict__.update(state)
        self._ncalled["__setstate__"] += 1
        return

    def __init__(self, display_handle):
        self._ncalled = {
            "__getstate__": 0,
            "__setstate__": 0,
            "_update_hostnames": 0,
            "_extract_host": 0,
            "_addr_to_hostname": 0,
            "open": 0,
            "process": 0,
            "close": 0
        }        
        self._display_handle = display_handle
        self._update_hostnames()
    
    def _extract_host(self, line):
        self._ncalled["_extract_host"] += 1
        sline = line.rstrip().split("\t")
        return (sline[0], sline[-1])

    def _update_hostnames(self):
        self._ncalled["_update_hostnames"] += 1
        with open("/etc/lx2-hosts", "r") as f:
            self._hostnames = dict(self._extract_host(line) for line in f)
        self._hostnames_updated = datetime.datetime.now()

    def _addr_to_hostname(self, addr):
        self._ncalled["_addr_to_hostname"] += 1
        if (datetime.datetime.now() - self._hostnames_updated).total_seconds() > 300:
            self._update_hostnames()
        return self._hostnames.get(addr, addr)

    def open(self, partition_id, epoch_id):
        self._ncalled["open"] += 1
        return True

    def process(self, row):
        self._ncalled["process"] += 1
        record = row.asDict(True)
        dst_addr = record["layers"]["ip"]["ip_ip_dst"]
        dst_hostname = self._addr_to_hostname(dest_addr)
        qry_name = record["layers"]["dns"]["text_dns_qry_name"]
        self._display_handle.update("\t".join([dst_hostname, qry_name]))
        return True

    def close(self, error):
        self._ncalled["close"] += 1
        return True

    
try:
    with tempfile.TemporaryDirectory() as tempdir:

        packets = spark.readStream\
            .option("cleanSource", "delete")\
            .json(tempdir + "/*.jsonl", schema=schema)\
            .withColumn("time", func.to_timestamp("timestamp"))

        record_loop = asyncio.new_event_loop()
        record_handle = record_loop.run_in_executor(
            concurrent.futures.ThreadPoolExecutor(max_workers=1),
            record_packets, 
            tempdir
        )

        display_handle = IPython.display.display(display_id=True)
        dns_operator = DnsRecordOperator(display_handle)

        def stream_results(df, epoch_id):
            def parse_row(row):
                record = row.asDict(True)
                dst_addr = record["layers"]["ip"]["ip_ip_dst"]
                dst_hostname = dns_operator._addr_to_hostname(dest_addr)
                qry_name = record["layers"]["dns"]["text_dns_qry_name"]
                return {
                    "dst_hostname": dst_hostname,
                    "qry_name": qry_name
                }
            
            out = pd.DataFrame(
                df.rdd.map(parse_row).collect()
            )
            display_handle.update(out)
            pass
        
        dns = packets.filter(func.col("layers.dns.text_dns_qry_name").isNotNull())
    
#        stream_query = dns.withWatermark("time", "2 minutes")\
#            .writeStream\
#            .outputMode("append")\
#            .foreach(dns_operator)\
#            .start()

        stream_query = dns.withWatermark("time", "2 minutes")\
            .writeStream\
            .outputMode("append")\
            .foreachBatch(stream_results)\
            .start()
    

#        record_loop.run_forever()
        
finally:
    pass
#    record_handle.cancel()
#    stream_query.stop()
# -

tempdir

json.dumps(dns_operator)



StarcoderdataJupyterScriptsDedupFiltered