标签: distributed-cache

我正在尝试将 a 存储TreeSet在 a 上，DistributedCache以供 Hadoop 映射缩减作业使用。到目前为止，我已经将文件从 HDFS 添加到DistributedCache：

Configuration conf = new Configuration();
DistributedCache.addCacheFile(new URI("/my/cache/path"), conf);
Job job = new Job(conf, "my job");
// Proceed with remainder of Hadoop map-reduce job set-up and running

如何有效地将 TreeSet（我已经在此类中构建）添加到我要添加到 DistributedCache 的文件中？我应该使用 Java 的本机序列化以某种方式将其序列化到文件中吗？

请注意，TreeSet 在启动 Map-Reduce 作业的主类中构建一次。TreeSet 永远不会被修改，我只是希望每个映射器都可以只读访问此 TreeSet，而不必一遍又一遍地重建它。

我正在尝试实现reduce side join,并使用mapfile reader查找分布式缓存,但是在stderr中检查时它没有查找值,它显示以下错误,lookupfile文件已经存在于hdfs中,并且似乎正确加载进入缓存,如stdout中所示.

java.lang.IllegalArgumentException:Wrong FS:file:/ app/hadoop/tmp/mapred/local/taskTracker/distcache/-8118663285704962921_-1196516983_170706299/localhost/input/delivery_status/DeliveryStatusCodes/data,expected:hdfs:// localhost:9000 org.apache.hadoop.fs.FileSystem.checkPath(FileSystem.java:390)位于org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus的org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:140) (DistributedFileSystem.java:554)atg.apache.hadoop.fs.FileSystem.getLength(FileSystem.java:816)org.apache.hadoop.io.SequenceFile $ Reader.(SequenceFile.java:1479)org.apache .hadoop.io.SequenceFile $ Reader.(SequenceFile.java:1474)org.apache.hadoop.io.MapFile $ Reader.createDataFileReader(MapFile.java:302)at org.apache.hadoop.io.MapFile $ Reader.打开(MapFile.java:284)org.apache.hadoop.io.MapFile $ Reader.(MapFile.java:273)org.apache.hadoop.io.MapFile $ Reader.(MapFile.java:260)at org .apache.hadoop.io.MapFile $读卡器(MapFile.java:253)在mr_poc.reducerrsj.initializeDepartmentsMap(reducerrsj.java:59)mr_poc.reducerrsj.setup(reducerrsj.java:42)atg.apache.hadoop.mapreduce.Reducer.run(Reducer.java:174)at org.位于org.apache.hadoop.mapred.Child $ 4.run的org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:418)中的apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:649) .java:255)at java.security.AccessController.doPrivileged(Native Method)at javax.security.auth.Subject.doAs(Subject.java:416)at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java) :1190)org.apache.hadoop.mapred.Child.main(Child.java:249)mr_poc.reducerrsj.reduce(reducerrsj.java)mr_poc.reducerrsj.buildOutputValue(reducerrsj.java:83)中的java.lang.NullPointerException :127)位于org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.)的org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:177)中的mr_poc.reducerrsj.reduce(reducerrsj.java:1). java:649)at org.apache.hadoop.mapred.ReduceTask.run(Reduc)eTask.java:418)atg.apache.hadoop.mapred.Child $ 4.run(Child.java:255)at java.security.AccessController.doPrivileged(Native Method)at javax.security.auth.Subject.doAs(Subject .java:416)在org.apache.hadoop.security.

这是我的驱动程序代码,

package mr_poc;

import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class driverrsj extends Configured implements Tool{

    @Override …

问题:我想缓存用户信息,以便我的所有应用程序都可以快速读取数据,但我只希望一个特定的应用程序能够写入此缓存.

我在AWS上,所以我遇到的一个解决方案是带有两个端口的memcached版本:一个端口只接受读命令,另一个端口接受读写操作.然后,我可以使用安全组来控制访问.

由于我在AWS上,如果有解决方案使用开箱即用的memcached或redis,那就太好了.

我有一个性能用例，通过该用例，我需要确定花费300毫秒以上的某些process()调用EntryProcessor。我尝试使用SlowOperationDetector以下配置。

    <!-- SlowOperation Detector Configuration -->
    <property name="hazelcast.slow.operation.detector.enabled">true</property>
    <property name="hazelcast.slow.operation.detector.stacktrace.logging.enabled">true</property>
    <property name="hazelcast.slow.operation.detector.log.purge.interval.seconds">60000</property>
    <property name="hazelcast.slow.operation.detector.log.retention.seconds">60000</property>
    <property name="hazelcast.slow.operation.detector.threshold.millis">300</property>

我给出了一个示例测试代码，该代码在内部睡眠1秒钟process()。

public static void main(String args[])
{
    Config cfg = null;
    try {
        cfg = new FileSystemXmlConfig("C:\\workarea\\hazelcast\\hazelcast-perf.xml");
    } catch (FileNotFoundException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(cfg);
    IMap<String, Employee> employeesMap = hazelcastInstance.getMap("anyMap");
    employeesMap.put("100", new Employee(100));
    SlowEntryProcessor slowEntryProcessor = new SlowEntryProcessor();
    employeesMap.executeOnKey("100", slowEntryProcessor);

}

static public class SlowEntryProcessor implements EntryProcessor<String, Employee> …

我最近开始在 dotnet core web api 上使用 IDistributedCache 的 sql 版本。

您将如何删除/使特定用户的一组密钥无效？

即：我构建了遵循此约定的密钥：

/users/{userId}/Key1
/users/{userId}/Key2
/users/{userId}/Section/Key3

我找不到任何方法来删除以下开头的所有键：/users/{userId}

如何一次从 IDistributedCache 中删除一项以上？