Используйте UDF для возврата списка, используя два столбца в кадре данных - PullRequest
Новая
       22

Используйте UDF для возврата списка, используя два столбца в кадре данных

0 голосов
/ 20 декабря 2018

У меня есть датафрейм dns (строка) и ip-адрес (строка).Я хотел бы использовать UDF для применения созданной мной функции python, которая ищет отдельные / уникальные днс и соотносит их с числом ips, с которым оно совпадает.Наконец, он выведет эту информацию в список.Конечным результатом является то, что UDF берет фрейм данных и возвращает список. 

#creating sample data
from pyspark.sql import Row
l = [('pipe.skype.com','172.25.132.26'),('management.azure.com','172.25.24.57'),('pipe.skype.com','172.11.128.10'),('management.azure.com','172.16.12.22'),('www.google.com','172.26.51.144'),('collector.exceptionless.io','172.22.2.21')]
rdd = sc.parallelize(l)
data = rdd.map(lambda x: Row(dns_host=x[0], src_ipv4=x[1]))
data_df = sqlContext.createDataFrame(data)

def beaconing_aggreagte(df):
  """Loops through unique hostnames and correlates them to unique src ip. If an individual hostname has less than 5 unique source ip connection, moves to the next step"""
  dns_host = df.select("dns_host").distinct().rdd.flatMap(lambda x: x).collect()
  HIT_THRESHOLD = 5
  data = []
  for dns in dns_host:
    dns_data =[]
    testing = df.where((f.col("dns_host") == dns)).select("src_ipv4").distinct().rdd.flatMap(lambda x: x).collect()
    if 0 < len(testing) <= 5: #must have less than 5 unique src ip for significance 
      dns_data.append(dns)
      data.append([testing,dns_data])
      print([testing,dns_data])
  return data

Я думаю, возможно, что моя схема неверна 

#Expected return from function: [[['172.25.24.57','172.16.12.22'],[management.azure.com]],..]
array_schema = StructType([
    StructField('ip', ArrayType(StringType()), nullable=False),
    StructField('hostname', ArrayType(StringType()), nullable=False)
]) 

testing_udf_beaconing_aggreagte = udf(lambda z: beaconing_aggreagte(z), array_schema)
df_testing = testing_df.select('*',testing_udf_beaconing_aggreagte(array('dns_host','src_ipv4')))
df_testing.show()

Эта ошибка: 

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1248.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1248.0 (TID 3846823, 10.139.64.23, executor 13): org.apache.spark.api.python.PythonException: Traceback (most recent call last):

Моя конечная цель - взять df и вернуть список в формате [[[list of ips], [dns_host]], ...].Я пытаюсь использовать UDF для распараллеливания операций над кластером вместо одного исполнителя.

1 Ответ

0 голосов
/ 21 декабря 2018

Группа должна быть в состоянии достичь этого.Используйте агрегат, чтобы собрать все IP-адреса, а затем подсчитать размер списка.Затем вы можете отфильтровать строки размером> 5 

from pyspark.sql.functions import *
from pyspark.sql import Row
l = [('pipe.skype.com','172.25.132.26'),('management.azure.com','172.25.24.57'),('pipe.skype.com','172.11.128.10'),('management.azure.com','172.16.12.22'),('www.google.com','172.26.51.144'),('collector.exceptionless.io','172.22.2.21')]
rdd = sc.parallelize(l)
data = rdd.map(lambda x: Row(dns_host=x[0], src_ipv4=x[1]))
data_df = sqlContext.createDataFrame(data)

data_df2 = data_df.groupby("dns_host").agg(F.collect_list("src_ipv4").alias("src_ipv4_list"))\
                  .withColumn("ip_count",F.size("src_ipv4_list"))\
                  .filter(F.col("ip_count") <= 5)\
                  .drop("ip_count")
data_df2.show(20,False)

Вывод: 

+--------------------------+------------------------------+
|dns_host                  |src_ipv4_list                 |
+--------------------------+------------------------------+
|pipe.skype.com            |[172.25.132.26, 172.11.128.10]|
|collector.exceptionless.io|[172.22.2.21]                 |
|www.google.com            |[172.26.51.144]               |
|management.azure.com      |[172.25.24.57, 172.16.12.22]  |
+--------------------------+------------------------------+
...