import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
private static void loadSparseFloatPartition(SparseFloatModel model, FSDataInputStream input,
    ModelPartitionMeta partMeta) throws IOException {
  int rowNum = input.readInt();
  int rowId = 0;
  int nnz = 0;
  int totalNNZ = 0;
  Int2FloatOpenHashMap row = null;
  for (int i = 0; i < rowNum; i++) {
    rowId = input.readInt();
    nnz = input.readInt();
    totalNNZ = (int) (nnz * (model.col) / (partMeta.getEndCol() - partMeta.getStartCol()));
    row = model.getRow(rowId, partMeta.getPartId(), totalNNZ);
    for (int j = 0; j < nnz; j++) {
      row.put(input.readInt(), input.readFloat());
    }
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Load dense double model to int->float maps
 *
 * @param modelDir model save directory path
 * @return model data
 */
public static Int2FloatOpenHashMap[] loadToFloatMaps(String modelDir, Configuration conf)
    throws IOException {
  // Load model meta
  ModelFilesMeta meta = getMeta(modelDir, conf);

  // Check row type
  if (meta.getRowType() != SPARSE_FLOAT) {
    throw new IOException("model row type is not sparse float, you should check it");
  }

  // Load model
  SparseFloatModel model = new SparseFloatModel(meta.getRow(), meta.getCol());
  loadModel(modelDir, model, meta, conf);

  return model.getModel();
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
private static void convertSparseFloatModel(Configuration conf, FSDataOutputStream output,
  String modelInputDir, ModelLineConvert lineConvert) throws IOException {
  Int2FloatOpenHashMap[] data = ModelLoader.loadToFloatMaps(modelInputDir, conf);
  for(int i = 0; i < data.length; i++) {
    Int2FloatOpenHashMap row = data[i];
    data[i] = null;
    if(row == null) {
      continue;
    }

    lineConvert.convertRowIndex(output, i);
    int [] indexes = row.keySet().toIntArray();
    float [] values = row.values().toFloatArray();
    row = null;
    Sort.quickSort(indexes, values, 0, indexes.length - 1);
    for(int j = 0; j < indexes.length; j++) {
      lineConvert.convertFloat(output, indexes[j], values[j]);
    }
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Test of sortByValue method, of class Utils.
 */
@org.junit.Test
public void testSortByValue() {
    System.out.println("sortByValue");
    Int2FloatOpenHashMap map = new Int2FloatOpenHashMap();
    map.put(3, 0.5f);
    map.put(2, 0.6f);
    map.put(4, 0.4f);
    map.put(1, 0.8f);
    map.put(5, 0f);
    Integer[] arrayResult = new Integer[map.size()];
    arrayResult = Utils.sortByValue(map, true).keySet().toArray(arrayResult);
    Integer[] correctResult = new Integer[]{1,2,3,4,5};
    assertArrayEquals(arrayResult, correctResult);     
            
    arrayResult = Utils.sortByValue(map, false).keySet().toArray(arrayResult);
    correctResult = new Integer[]{5,4,3,2,1};
    assertArrayEquals(arrayResult, correctResult);     
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
private void replayTrain(@Nonnull final ByteBuffer buf) {
    final int itemI = buf.getInt();
    final int knnSize = buf.getInt();

    final Int2ObjectMap<Int2FloatMap> knnItems = new Int2ObjectOpenHashMap<>(1024);
    final IntSet pairItems = new IntOpenHashSet();
    for (int i = 0; i < knnSize; i++) {
        int user = buf.getInt();
        int ruSize = buf.getInt();
        Int2FloatMap ru = new Int2FloatOpenHashMap(ruSize);
        ru.defaultReturnValue(0.f);

        for (int j = 0; j < ruSize; j++) {
            int itemK = buf.getInt();
            pairItems.add(itemK);
            float ruk = buf.getFloat();
            ru.put(itemK, ruk);
        }
        knnItems.put(user, ru);
    }

    for (int itemJ : pairItems) {
        train(itemI, knnItems, itemJ);
    }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Nonnull
private static Int2FloatMap int2floatMap(@Nonnull final Map<?, ?> map,
        @Nonnull final PrimitiveObjectInspector keyOI,
        @Nonnull final PrimitiveObjectInspector valueOI) {
    final Int2FloatMap result = new Int2FloatOpenHashMap(map.size());
    result.defaultReturnValue(0.f);

    for (Map.Entry<?, ?> entry : map.entrySet()) {
        float v = PrimitiveObjectInspectorUtils.getFloat(entry.getValue(), valueOI);
        if (v == 0.f) {
            continue;
        }
        int k = PrimitiveObjectInspectorUtils.getInt(entry.getKey(), keyOI);
        result.put(k, v);
    }

    return result;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Constructor
 *
 * @param service Service worker
 * @param combiner Message combiner
 */
public IntFloatMessageStore(
    CentralizedServiceWorker<IntWritable, ?, ?> service,
    Combiner<IntWritable, FloatWritable> combiner) {
  this.service = service;
  this.combiner = combiner;

  map = new Int2ObjectOpenHashMap<Int2FloatOpenHashMap>();
  for (int partitionId : service.getPartitionStore().getPartitionIds()) {
    Partition<IntWritable, ?, ?> partition =
        service.getPartitionStore().getPartition(partitionId);
    Int2FloatOpenHashMap partitionMap =
        new Int2FloatOpenHashMap((int) partition.getVertexCount());
    map.put(partitionId, partitionMap);
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Get a model row
 *
 * @param rowId row index
 * @param partId partition index
 * @return model row
 */
public Int2FloatOpenHashMap getRow(int rowId, int partId) {
  synchronized (this) {
    if (tempModel.get(rowId) == null) {
      tempModel.put(rowId, new HashMap<>());
      tempModel.get(rowId).put(partId, new Int2FloatOpenHashMap());
    } else {
      if (tempModel.get(rowId).get(partId) == null) {
        tempModel.get(rowId).put(partId, new Int2FloatOpenHashMap());
      }
    }

    return tempModel.get(rowId).get(partId);
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
public static Int2FloatOpenHashMap loadSparseFloatRowFromPartition(FSDataInputStream input,
    ModelPartitionMeta partMeta, int rowId) throws IOException {
  RowOffset rowOffset = partMeta.getRowMetas().get(rowId);
  input.seek(rowOffset.getOffset());
  Preconditions.checkState (input.readInt() == rowId);
  int num = input.readInt();
  Int2FloatOpenHashMap row = new Int2FloatOpenHashMap();
  for (int i = 0; i < num; i++) {
    row.put(input.readInt(), input.readFloat());
  }
  return row;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * init the vector by setting the dim and capacity
 *
 * @param dim vector dimension
 * @param capacity map capacity
 */
public SparseFloatVector(int dim, int capacity) {
  super();
  if(capacity > 0){
    this.hashMap = new Int2FloatOpenHashMap(capacity);
  } else {
    this.hashMap = new Int2FloatOpenHashMap(INIT_SIZE);
  }
  this.dim = dim;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
public void deserialize(ByteBuf buf) {
  int dim = buf.readInt();
  int length = buf.readInt();
  Int2FloatOpenHashMap data = new Int2FloatOpenHashMap(length);
  IntStream.range(0,length).forEach(i-> data.put(buf.readInt(), buf.readFloat()));
  this.dim = dim;
  this.hashMap = data;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Merge this sparse float vector split to a map
 * @param indexToValueMap a index->value map
 */
public void mergeTo(Int2FloatOpenHashMap indexToValueMap) {
  try {
    lock.readLock().lock();
    indexToValueMap.putAll(hashMap);
  } finally {
    lock.readLock().unlock();
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * @deprecated Not needed in the new implementation. Kept, in case it will be useful in the future. 
 * Returns the neighborSim of each entity pair. 
 * @param valueSims an RDD with the top-K value sims for each entity, in the form key: eid, value: (candidateMatch cId, value_sim(eId,cId))
 * @param inNeighbors_BV
 * @return the neighborSim of each entity pair, where entity pair is the key and neighborSim is the value
 */
private JavaPairRDD<Tuple2<Integer, Integer>, Float> getNeighborSims(JavaPairRDD<Integer,Int2FloatOpenHashMap> valueSims, Broadcast<Map<Integer,IntArrayList>> inNeighbors_BV) {
   
   return valueSims.flatMapToPair(x->{
        int eId = x._1();
        IntArrayList eInNeighbors = inNeighbors_BV.value().get(eId);
        
        List<Tuple2<Tuple2<Integer,Integer>, Float>> partialNeighborSims = new ArrayList<>(); //key: (negativeEid, positiveEid), value: valueSim(outNeighbor(nEid),outNeighbor(pEid))
        if (eInNeighbors == null) {
            return partialNeighborSims.iterator(); //empty
        }
        for (Map.Entry<Integer, Float> eIdValueCandidates : x._2().entrySet()) { 
            IntArrayList inNeighborsOfCandidate = inNeighbors_BV.value().get(eIdValueCandidates.getKey());
            if (inNeighborsOfCandidate == null) {
                continue; //go to next candidate match. this one does not have in-neighbors
            }
            Float tmpNeighborSim = eIdValueCandidates.getValue();
            for (Integer inNeighborOfCandidate : inNeighborsOfCandidate) { //for each in-neighbor of the candidate match of the current entity                    
                for (Integer eInNeighbor : eInNeighbors) {  //for each in-neighbor of the current entity
                    if (eId < 0) 
                        partialNeighborSims.add(new Tuple2<>(new Tuple2<>(eInNeighbor, inNeighborOfCandidate), tmpNeighborSim));
                    else 
                        partialNeighborSims.add(new Tuple2<>(new Tuple2<>(inNeighborOfCandidate, eInNeighbor), tmpNeighborSim));
                }
            }
        }
        
        return partialNeighborSims.iterator();
    })
    .reduceByKey((w1, w2) -> Math.max(w1, w2)); //for each entity pair, neighborSim = max value sim of its pairs of out-neighbors        
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
public FMIntFeatureMapModel(@Nonnull FMHyperParameters params) {
    super(params);
    this._w0 = 0.f;
    this._w = new Int2FloatOpenHashMap(DEFAULT_MAPSIZE);
    _w.defaultReturnValue(0.f);
    this._V = new Int2ObjectOpenHashMap<float[]>(DEFAULT_MAPSIZE);
    this._minIndex = 0;
    this._maxIndex = 0;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
protected PredictionModel createModel() {
    this._w0 = 0.f;
    this._w1 = new Int2FloatOpenHashMap(16384);
    _w1.defaultReturnValue(0.f);
    this._w2 = new Int2FloatOpenHashMap(16384);
    _w2.defaultReturnValue(0.f);
    this._w3 = new Int2FloatOpenHashMap(16384);
    _w3.defaultReturnValue(0.f);

    return null;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
/**
 * Create a SparseFloatVector consisting of double values according to the
 * specified mapping of indices and values.
 *
 * @param values the values to be set as values of the real vector
 * @param dimensionality the dimensionality of this feature vector
 * @throws IllegalArgumentException if the given dimensionality is too small
 *         to cover the given values (i.e., the maximum index of any value not
 *         zero is bigger than the given dimensionality)
 */
public SparseFloatVector(Int2FloatOpenHashMap values, int dimensionality) throws IllegalArgumentException {
  if(values.size() > dimensionality) {
    throw new IllegalArgumentException("values.size() > dimensionality!");
  }

  this.indexes = new int[values.size()];
  this.values = new float[values.size()];
  // Import and sort the indexes
  {
    ObjectIterator<Int2FloatMap.Entry> iter = values.int2FloatEntrySet().fastIterator();
    for(int i = 0; iter.hasNext(); i++) {
      this.indexes[i] = iter.next().getIntKey();
    }
    Arrays.sort(this.indexes);
  }
  // Import the values accordingly
  {
    for(int i = 0; i < values.size(); i++) {
      this.values[i] = values.get(this.indexes[i]);
    }
  }
  this.dimensionality = dimensionality;
  final int maxdim = getMaxDim();
  if(maxdim > dimensionality) {
    throw new IllegalArgumentException("Given dimensionality " + dimensionality + " is too small w.r.t. the given values (occurring maximum: " + maxdim + ").");
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
public void addPartitionMessages(int partitionId,
    ByteArrayVertexIdMessages<IntWritable, FloatWritable> messages) throws
    IOException {
  IntWritable reusableVertexId = new IntWritable();
  FloatWritable reusableMessage = new FloatWritable();
  FloatWritable reusableCurrentMessage = new FloatWritable();

  Int2FloatOpenHashMap partitionMap = map.get(partitionId);
  synchronized (partitionMap) {
    ByteArrayVertexIdMessages<IntWritable,
        FloatWritable>.VertexIdMessageIterator
        iterator = messages.getVertexIdMessageIterator();
    while (iterator.hasNext()) {
      iterator.next();
      int vertexId = iterator.getCurrentVertexId().get();
      float message = iterator.getCurrentMessage().get();
      if (partitionMap.containsKey(vertexId)) {
        reusableVertexId.set(vertexId);
        reusableMessage.set(message);
        reusableCurrentMessage.set(partitionMap.get(vertexId));
        combiner.combine(reusableVertexId, reusableCurrentMessage,
            reusableMessage);
        message = reusableCurrentMessage.get();
      }
      partitionMap.put(vertexId, message);
    }
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
public Iterable<FloatWritable> getVertexMessages(
    IntWritable vertexId) throws IOException {
  Int2FloatOpenHashMap partitionMap = getPartitionMap(vertexId);
  if (!partitionMap.containsKey(vertexId.get())) {
    return EmptyIterable.get();
  } else {
    return Collections.singleton(
        new FloatWritable(partitionMap.get(vertexId.get())));
  }
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
public Iterable<IntWritable> getPartitionDestinationVertices(
    int partitionId) {
  Int2FloatOpenHashMap partitionMap = map.get(partitionId);
  List<IntWritable> vertices =
      Lists.newArrayListWithCapacity(partitionMap.size());
  IntIterator iterator = partitionMap.keySet().iterator();
  while (iterator.hasNext()) {
    vertices.add(new IntWritable(iterator.nextInt()));
  }
  return vertices;
}
 

import it.unimi.dsi.fastutil.ints.Int2FloatOpenHashMap; //导入依赖的package包/类
@Override
public void writePartition(DataOutput out,
    int partitionId) throws IOException {
  Int2FloatOpenHashMap partitionMap = map.get(partitionId);
  out.writeInt(partitionMap.size());
  ObjectIterator<Int2FloatMap.Entry> iterator =
      partitionMap.int2FloatEntrySet().fastIterator();
  while (iterator.hasNext()) {
    Int2FloatMap.Entry entry = iterator.next();
    out.writeInt(entry.getIntKey());
    out.writeFloat(entry.getFloatValue());
  }
}