Untitled

	logger.info("selectInnerNetworkDataTraffic..");
		Dataset<Row> selectInner = GetNetworkDataTraffic //
		.groupBy(
				col(NetworkDataTrafficModel.MSISDN_VALUE.columnName()),
				col(NetworkDataTrafficModel.TYPE_ALLOCATION_CODE.columnName()),
				col(NetworkDataTrafficModel.CELL_GLOBAL_IDENTITY_CODE.columnName()),
				col(NetworkDataTrafficModel.TRANSACTION_DATE.columnName())
		)
		.agg(count("*").alias("cnt"))
		.select(
				col(NetworkDataTrafficModel.TYPE_ALLOCATION_CODE.columnName()),
				col(NetworkDataTrafficModel.CELL_GLOBAL_IDENTITY_CODE.columnName()),
				col(NetworkDataTrafficModel.TRANSACTION_DATE.columnName()),
				sum(col(NetworkDataTrafficModel.TOTAL_DATA_VOLUME_UPLOAD_BYTES.columnName()))
					.plus(sum(col(NetworkDataTrafficModel.TOTAL_DATA_VOLUME_DOWNLOAD_BYTES.columnName()))).alias("total_data_volume_bytes_radcom"),
				sum(col(NetworkDataTrafficModel.UPLOAD_TOTAL_PROCEDURE_COUNT.columnName())).alias("upload_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.DOWNLOAD_TOTAL_PROCEDURE_COUNT.columnName())).alias("download_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.MEAN_ROUND_TRIP_TIME_DURATION_COUNT.columnName())).alias("mean_round_trip_time_duration_count"),
				sum(col(NetworkDataTrafficModel.MEAN_UPLOAD_THROUGHPUT_RATE.columnName())).alias("mean_upload_throughput_rate"),
				sum(col(NetworkDataTrafficModel.MEAN_DOWNLOAD_THROUGHPUT_RATE.columnName())).alias("mean_download_throughput_rate"),
				sum(col(NetworkDataTrafficModel.ROUND_TRIP_TIME_TOTAL_PROCEDURE_COUNT.columnName())).alias("round_trip_time_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.TOTAL_PROCEDURE_COUNT.columnName())).alias("total_procedure_count"),
				col(NetworkDataTrafficModel.MSISDN_VALUE.columnName()).alias("msisdn_value")
		);
		selectInner = selectInner.drop(col("cnt"));


		logger.info("GetAggNetworkDataTraffic..");
		Dataset<Row> AggNetworkDataTraffic = selectInner
		.groupBy(
				col(NetworkDataTrafficModel.MSISDN_VALUE.columnName()),
				col(NetworkDataTrafficModel.TYPE_ALLOCATION_CODE.columnName()),
				col(NetworkDataTrafficModel.CELL_GLOBAL_IDENTITY_CODE.columnName()),
				col(NetworkDataTrafficModel.TRANSACTION_DATE.columnName())
		)
		.agg(count("*").alias("cnt"))
		.select(
				col(NetworkDataTrafficModel.TYPE_ALLOCATION_CODE.columnName()).alias("type_allocation_code"),
				col(NetworkDataTrafficModel.CELL_GLOBAL_IDENTITY_CODE.columnName()).alias("cell_global_identity_code"),
				col(NetworkDataTrafficModel.TRANSACTION_DATE.columnName()).alias("transaction_date"),
				col(NetworkDataTrafficModel.TYPE_ALLOCATION_CODE.columnName()).alias("type_allocation_code"),
				sum(col(NetworkDataTrafficModel.TOTAL_DATA_VOLUME_BYTES_RADCOM.columnName())).alias("total_data_volume_bytes_radcom"),
				sum(col(NetworkDataTrafficModel.UPLOAD_TOTAL_PROCEDURE_COUNT.columnName())).alias("upload_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.DOWNLOAD_TOTAL_PROCEDURE_COUNT.columnName())).alias("download_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.MEAN_ROUND_TRIP_TIME_DURATION_COUNT.columnName())).alias("mean_round_trip_time_duration_count"),
				sum(col(NetworkDataTrafficModel.ROUND_TRIP_TIME_TOTAL_PROCEDURE_COUNT.columnName())).alias("round_trip_time_total_procedure_count"),
				sum(col(NetworkDataTrafficModel.MEAN_UPLOAD_THROUGHPUT_RATE.columnName())).alias("mean_upload_throughput_rate"),
				sum(col(NetworkDataTrafficModel.MEAN_DOWNLOAD_THROUGHPUT_RATE.columnName())).alias("mean_download_throughput_rate"),
				sum(col(NetworkDataTrafficModel.TOTAL_PROCEDURE_COUNT.columnName())).alias("total_procedure_count"),
				col(NetworkDataTrafficModel.MSISDN_VALUE.columnName()).alias("msisdn_value")
		);
		AggNetworkDataTraffic = AggNetworkDataTraffic.drop(col("cnt"));