# ---
# 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
# ---

from pyspark import SparkConf
from pyspark.sql import SparkSession, SQLContext
from pyspark.sql import functions as F
from functools import reduce
from pyspark.sql.window import Window
from pyspark.sql.types import IntegerType
#Python imports
import sys
sys.path.append("/home/svanhmic/workspace/Python/Erhvervs/src/cvr")
import GetNextJsonLayer
import os
path = "/home/svanhmic/workspace/Python/Erhvervs/data/cdata"
#paths


# #%ls /home/svanhmic/workspace/Python/Erhvervs/data/cdata 
# %env
# #%pdb
from graphframes import graphframe



df = sqlContext.read.json(path+"/AlleDeltager.json")
deltagerCols = GetNextJsonLayer.getNextSchemaLayer(df.schema,"deltager")

deltagerCols
df.printSchema()

# +
deltagerAttributterDf = GetNextJsonLayer.createNextLayerTable(df,["enhedsNummer","enhedstype","navne"],"attributter","deltager")
holdOutCol = ["vaerdier"]
vaerdiCols = [i for i in deltagerAttributterDf.columns if i not in holdOutCol]

deltagerAttributterWithValuesDf = (GetNextJsonLayer
                                   .createNextLayerTable(deltagerAttributterDf,vaerdiCols,holdOutCol[0])
                                   .drop("periode")
                                   .drop("sidstOpdateret")
                                  )
del holdOutCol[0]
holdOutCol.append("navne")
del vaerdiCols
vaerdiCols = [i for i in deltagerAttributterWithValuesDf.columns if i not in holdOutCol]
withNamesAttributsDf = (GetNextJsonLayer
                        .createNextLayerTable(deltagerAttributterWithValuesDf,vaerdiCols,holdOutCol[0])
                        .drop("sidstOpdateret")
                       )
#deltagerAttWithValsAndPeriodDf = GetNextJsonLayer.expandSubCols(df=deltagerAttributterWithValuesDf,mainColumn="periode")
#deltagerAttributterDf.columns


# +
#withNamesAttributsDf.show(truncate=False)
# -

holdtoCols = ["enhedsNummer","enhedstype","navne","stilling"]
delVirkRelDf = (GetNextJsonLayer
                .createNextLayerTable(df,holdtoCols,"virksomhedSummariskRelation","deltager")
                .withColumnRenamed(existing="enhedsNummer",new="personEnhedsNummer")
                .withColumnRenamed(existing="enhedstype",new="personEnhedstype")
                .withColumnRenamed(existing="stilling",new="personStilling")
                .withColumnRenamed(existing="navne",new="personNavne")
                .drop("organisationer")
               )

delVirkRelDf.show(20,truncate=False)
delVirkRelDf.printSchema()

# +
lenUdf = F.udf(lambda x: len(x),IntegerType())

delVirkRelDf.select("personNavne",lenUdf(F.col("personNavne")).alias("length")).orderBy(F.col("length").asc()).show()

# +
delCols = [i for i in delVirkRelDf.columns if i not in ["organisationer","virksomhed"]]
delVirkRelOrgDf =  (GetNextJsonLayer
                    .expandSubCols(delVirkRelDf,"virksomhed")
                    .drop("livsforloeb")
                    .drop(F.col("fejlRegistreret"))
                    .drop("virksomhedsstatus")
                    .drop("virksomhedsform")
                    .drop("sidstIndlaest")
                    .drop("status")
                    .drop("regNummer")
                    .withColumnRenamed(existing="enhedsNummer",new="firmaEnhedsNummer")
                    .withColumnRenamed(existing="navne",new="firmaNavne")
                    .withColumnRenamed(existing="enhedstype",new="firmaEnhedstype")
                    .cache()
                   )

holdOutCol = ["personNavne"]
namesCol = [i for i in delVirkRelOrgDf.columns if i not in holdOutCol]
delVirkRelWithPerNamDf = (GetNextJsonLayer
                          .createNextLayerTable(delVirkRelOrgDf,namesCol,holdOutCol[0])
                          .drop("sidstOpdateret")
                          .withColumnRenamed(existing="navn",new="personNavn")
                         )
del holdOutCol
del namesCol
holdOutCol = ["firmaNavne"]
namesCol = [i for i in delVirkRelWithPerNamDf.columns if i not in holdOutCol]
delVirkRelWithVirkNamDf = (GetNextJsonLayer
                           .createNextLayerTable(delVirkRelWithPerNamDf,namesCol,holdOutCol[0])
                           .drop("sidstOpdateret")
                           .drop("periode")
                           .withColumnRenamed(existing="navn",new="firmaNavn")
                          )

delVirkRelWithVirkNamDf.show()
# -

#create person to oraganisation edge 
personFirmaEdgeDf = (delVirkRelWithVirkNamDf.select(F.col("personEnhedsNummer").alias("src"),F.col("firmaEnhedsNummer").alias("dst")))

# +
personFirmaEdgeDf.show()


# -







allPersons = (virkSumRelationWithNamesDf
                  .select(F.col("enhedsNummer").alias("id")
                          ,F.col("navne_navn").alias("navn")
                          ,F.lit("Person").alias("type"))
                  .distinct()
                  .cache()
                 )
allOrgs = (virkSumRelationWithNamesDf
           .select(F.col("enhedsNummerOrganisation").alias("id")
                   ,F.col("orgNavn_navn").alias("navn")
                   ,F.lit("Organisation").alias("type"))
           .filter(F.col("id") !=0)
           .cache()
          )
allVerticesDf = allOrgs.unionAll(allPersons)

allPersons.count()
allOrgs.count()

personOrgGrsaphDf = graphframe.GraphFrame(allVerticesDf,personTilFirmaEdgeDf).cache()
groupedCols = [F.col("type")]
personOrgGraphDf.vertices.groupby(*groupedCols).count().show()

personOrgGraphDf.degrees.show()

personOrgGraphDf.outDegrees

personOrgGraphDegreesDf = personOrgGraphDf.vertices.join(personOrgGraphDf.outDegrees,(personOrgGraphDf.vertices["id"]==personOrgGraphDf.outDegrees["id"]),"inner")

personOrgGraphDegreesDf.show()



StarcoderdataJupyterScriptsDedupFiltered