import numpy as np

import pandas as pd

def get_distances(cores, data: np.ndarray, Numerical, Type, gamma) -> np.ndarray:

'''

距离计算函数

'''

m, n = data.shape

k = cores.shape[0]

d1 = (np.abs(np.repeat(data[:, Numerical], k, axis=0).reshape(m, k, len(Numerical)) - cores[:, Numerical]))*np.array(gamma)[Numerical]

distance_1 = np.sum(d1, axis=2) # ndarray(m, k)

distance_2 = np.sum((np.repeat(data[:, Type], k, axis=0).reshape(m, k, len(Type)) != cores[:, Type])*np.array(gamma)[Type], axis=2)

return distance_1 + distance_2

def get_new_cores(result, data, cores):

'''

重新找聚类中心

'''

k = cores.shape[0]

for i in range(k): # 遍历质心集

items = data[result == i] # 找出对应当前质心的子样本集

cores[i, Numerical] = np.mean(items[:, Numerical], axis=0) # 更新数据型数据核心

for num in Type:

cores[i, num] = np.argmax(np.bincount(items[:, num])) # 更新分类型数据核心

return cores

def k_prototypes(data, k, Numerical, Type, gamma):

"""

k-means聚类算法

k - int,指定分簇数量

data - ndarray(m, n)，m个样本的数据集，每个样本n个属性值

Numerical - list,数值型数据所在列

Type - list, 分类型数据所在的列

gamma - float, 分类属性权重。

- list, 若为list,保证 len(gamma)==len(Type),每个分类型数据的权重可单独设置

"""

m, n = data.shape # m：样本数量，n：每个样本的属性值个数

result = np.empty(m) # 生成容器，用于存放m个样本的聚类结果

cores = data[np.random.choice(np.arange(m), k, replace=False)] # 从m个数据样本中不重复地随机选择k个样本作为质心

while True: # 迭代计算

distance = get_distances(cores, data, Numerical, Type, gamma) # 计算距离矩阵

index_min = np.argmin(distance, axis=1) # 每个样本距离最近的质心索引序号

if (index_min == result).all(): # 如果样本聚类没有改变

return result, cores # 则返回聚类结果和质心数据

result[:] = index_min # 重新分类

cores = get_new_cores(result, data, cores) # 重新寻找聚类中心

data=pd.read_csv(r'/root/data_tmp/四川未成年人聚类.csv',header=0).values

# data数据共有8列，其中 0-5列的数据为分类型数据，6-7列为数值型数据。

Numerical=[6,7]

Type=[0,1,2,3,4,5]

gamma=[1,1,1,1,1,1,2/7,3/18] # 定义每项指标的占比

k=5 # 设置聚类

all_result={} # 用于保存运行结果，保存为json格式

# 运行10

for i in range(10):

result, cores = k_prototypes(data, k, Numerical, Type, gamma)

all_result[i]={}

all_result[i]['result']=result # 保存分类结果

all_result[i]['cores'] = cores # 保存分类的簇中心

cores_nums=[]

for one_k in range(k):

cores_nums.append(int(sum(result==one_k))) # 对数据进行转化，后续保存为json格式会报错

all_result[i]['cores_nums'] = cores_nums # 保存每个簇的数量

3.6 效果评价

import os

import sys

from pyspark import SparkConf

from pyspark.sql import SparkSession

# 如果当前代码文件运行测试需要加入修改路径，避免出现后导包问题

BASE_DIR = os.path.dirname(os.path.dirname(os.getcwd()))

sys.path.insert(0, os.path.join(BASE_DIR))

PYSPARK_PYTHON = "/opt/anaconda3/envs/pythonOnYarn/bin/python3.8"

# 当存在多个版本时，不指定很可能会导致出错

os.environ["PYSPARK_PYTHON"] = PYSPARK_PYTHON

os.environ["PYSPARK_DRIVER_PYTHON"] = PYSPARK_PYTHON

os.environ["HADOOP_USER_NAME"]="hdfs"

spark=SparkSession.builder.master("yarn").appName("chen") \

.config("spark.yarn.queue","q2") \

.config("spark.executor.instances",80) \

.config("spark.executor.cores",8) \

.config("spark.executor.memory","4g") \

.enableHiveSupport() \

.getOrCreate()

sc=spark.sparkContext

# 基本参数设置

Numerical=[6,7]

Type=[0,1,2,3,4,5]

k=5

broadcast=sc.broadcast({'k':k,'Num_list':Numerical,'Type_list':Type,'gamma':gamma}) # 广播变量

# 基础数据处理

import pandas as pd

data=pd.read_csv(r'/root/聚类数据.csv',header=0).values

data_list = data.tolist()

data_tuple = [(i, tuple(data_list[i])) for i in range(len(data_list))] # 将数据转化为元组

rdd = sc.parallelize(data_tuple, 75).cache() # 将数据转化为rdd，分区数设为75，

broad_rdd = sc.broadcast(data_tuple) # 将元数据进行广播

for key,value in all_result.items(): # all_result 是由之前代码运行的结果

result=np.array(value['result']) # 转成ndarray格式

cores=np.array(value['cores'])

broadresult=sc.broadcast(tuple(result.astype(int).tolist())) # 将当前分类结果进行广播

def get_distance_rdd(one_rdd):

broadcast_value=broadcast.value

broadResult=broadresult.value

broadrdd=broad_rdd.value

def distance(rdd1,rdd2):

sum_=0

for i in range(len(rdd1[1])):

if i in broadcast_value['Num_list']:

sum_+=abs(rdd1[1][i]-rdd2[1][i])*broadcast_value['gamma'][i]

elif i in broadcast_value['Type_list']:

sum_+=0 if rdd1[1][i]==rdd2[1][i] else broadcast_value['gamma'][i]

return sum_

cluster_in=[0,0] # [簇内的距离之和，簇内最远距离]

cluster_out=[float('inf')]*broadcast_value['k']

for one in broadrdd:

dis_tmp=distance(one_rdd,one)

if broadResult[one[0]] == broadResult[one_rdd[0]]: # 同一簇内

cluster_in[0]+=dis_tmp

if dis_tmp>cluster_in[1]:

cluster_in[1]=dis_tmp

else:

if cluster_out[broadResult[one[0]]]>dis_tmp:

cluster_out[broadResult[one[0]]]=dis_tmp

return (broadResult[one_rdd[0]],cluster_in,cluster_out) # (所属类别,(簇内平均距离，簇内最远距离),(不同簇间的最近的距离))

res=rdd.map(get_distance_rdd).cache()

target=res.map(lambda x:(x[0],x[1]+x[2])).reduceByKey(lambda x,y:[x[0]+y[0],x[1] if x[1]>=y[1] else y[1]]+[x[i] if x[i]<=y[i] else y[i] for i in range(2,len(x))]).sortByKey(ascending=True).collect()

avg=[]

diam=[]

dmin=[]

for one in target:

avg.append(one[1][0]/(sum(result==one[0])**2))

diam.append(one[1][1])

dmin.append(one[1][2:])

dcen=get_distances(cores, cores, Numerical, Type, gamma).astype(int).tolist()

DBI_list=[-float('inf')]*k

for i in range(k):

for j in range(k):

if i!=j:

tmp=(avg[i]+avg[j])/dcen[i][j]

if tmp>DBI_list[i]:

DBI_list[i]=tmp

DBI=sum(DBI_list)/k

DI=min([min(one) for one in dmin])/max(diam)

all_result[key]['avg']=avg

all_result[key]['diam'] = diam

all_result[key]['dmin'] = dmin

all_result[key]['dcen'] = dcen

all_result[key]['DBI'] = DBI

all_result[key]['DI'] = DI

# 将运行结果保存为json文件

import json

def dic_to_json(dic,save_path):

tf=open(save_path,"w")

json.dump(dic,tf)

tf.close()

dic_to_json(all_result,r'/root/k_prototypes_k_5结果.json')